""" Predictive Maintenance Light Lokales Mini-HMI Dashboard fuer Motoren und Antriebe. Nur Python-Standardbibliothek, HTML, CSS und JavaScript. """ import csv import json import os import random import sys import time from datetime import datetime import http.server import socketserver import threading import urllib.parse import webbrowser SCRIPT_ORDNER = os.path.dirname(os.path.abspath(__file__)) DASHBOARD_DATEI = "dashboard.html" STYLE_DATEI = "style.css" DATEN_DATEI = "dashboard_data.json" HISTORIE_DATEI = "motor_history.csv" ALARM_HISTORIE_DATEI = "alarm_history.csv" QUITTIERUNGEN_DATEI = "acknowledged_alarms.json" REPARATUR_HISTORIE_DATEI = "repair_history.csv" SETTINGS_DATEI = "settings.json" SIMULATION_STATE_DATEI = "simulation_state.json" ACTIVE_ALARM_STATE_DATEI = "active_alarm_state.json" SCHICHTUEBERGABE_DATEI = "schichtuebergabe.txt" TAGESBERICHT_DATEI = f"tagesbericht_{datetime.now().strftime('%Y-%m-%d')}.txt" SERVER_PORT = 8000 AKTUALISIERUNG_SEKUNDEN = 2 STROMPREIS_EURO_KWH = 0.25 JSON_SCHREIB_LOCK = threading.RLock() MESSRUNDEN_LOCK = threading.RLock() SIMULATIONS_MODI = [ "standard", "normal", "auffaellig", "kritisch", "pumpe_warnung", "luefter_kritisch", "sensorfehler", "versorgungsfehler", "400v_fehler", "fu_stoerung", "materialstau", "wiederholfehler", "reparatur_demo", "pruefprotokoll_erforderlich", "meisterfreigabe", "cnc_stoerung", "saege_stoerung", "reinigung_stoerung", "roboter_stoerung", "pruefung_stoerung", "achse_schwergaengig", "spindel_lager_warnung", "reinigung_druckverlust", "sicherheit_not_aus", "medien_luftdruck_niedrig", ] ERSATZTEILBESTAND = { "FB1-LAG-6205-2RS": { "bestand": 2, "mindestbestand": 1, "lagerort": "Instandhaltung Regal A2", "status": "auf Lager", }, "FB1-FU-5500": { "bestand": 0, "mindestbestand": 1, "lagerort": "nicht vorhanden", "status": "Bestellung empfohlen", }, "FB1-SEN-VIB-020": { "bestand": 1, "mindestbestand": 1, "lagerort": "Sensorik Schrank S1", "status": "auf Lager", }, "FB1-SEN-IN-001": { "bestand": 1, "mindestbestand": 1, "lagerort": "Sensorik Schrank S1", "status": "auf Lager", }, "FB1-SEN-OUT-001": { "bestand": 1, "mindestbestand": 1, "lagerort": "Sensorik Schrank S1", "status": "auf Lager", }, } BETRIEBSARTEN = [ "automatik", "handbetrieb", "wartung", "stoerung", "gestoppt", ] PRAESENTATIONS_SCHRITTE = [ { "titel": "Anlage Grün", "modus": "normal", "betriebsart": "automatik", "betroffen": "Gesamte Linie 1 Metallbearbeitung", "auswirkung": "Alle Komponenten OK, Materialfluss aktiv, Produktion im Soll.", }, { "titel": "Materialeingang — BAND_1 aktiv", "modus": "normal", "betriebsart": "automatik", "betroffen": "BAND_1, Materialeingang", "auswirkung": "Metallblock wird angeliefert, Förderband 1 transportiert Material zur Sägestation.", }, { "titel": "Sägestation — Zersägen aktiv", "modus": "normal", "betriebsart": "automatik", "betroffen": "SAEGE_M1, SAEGE_M2, SAEGE_M3", "auswirkung": "Hauptsägeantrieb, Vorschub und Kühlung laufen im Normalbetrieb.", }, { "titel": "CNC-Frässtation — Bearbeitung läuft", "modus": "normal", "betriebsart": "automatik", "betroffen": "CNC_SP1, CNC_SP2, CNC_SP3, CNC_SP4", "auswirkung": "CNC-Spindeln aktiv, alle Achsen referenziert, Kühlschmierstoff OK.", }, { "titel": "CNC-Störung — Spindel 2 Alarm", "modus": "cnc_stoerung", "betriebsart": "automatik", "betroffen": "CNC_SP2, CNC-Station", "auswirkung": "CNC_SP2 Risikowert kritisch, Alarm aktiv, Produktion an CNC eingeschränkt.", }, { "titel": "Reinigung + Roboter KUKA_R1", "modus": "normal", "betriebsart": "automatik", "betroffen": "REIN_M1, REIN_M2, REIN_M3, KUKA_R1", "auswirkung": "Teile werden gereinigt, KUKA_R1 handhabt Werkstücke zur Reinigungsstation.", }, { "titel": "Schwarzkammer — Prüfstation", "modus": "normal", "betriebsart": "automatik", "betroffen": "Schwarzkammer, Prüfstation", "auswirkung": "Sichtprüfung und Maßkontrolle aktiv, alle Teile werden vermessen.", }, { "titel": "Versand — KUKA_R2 aktiv", "modus": "normal", "betriebsart": "automatik", "betroffen": "KUKA_R2, Versandstation", "auswirkung": "KUKA_R2 palettiert fertige Teile, Versand läuft planmäßig.", }, { "titel": "Diagnose — Sensorfehler", "modus": "sensorfehler", "betriebsart": "automatik", "betroffen": "Sensorik, Messkette, Datenqualität", "auswirkung": "Datenqualität sinkt, Signalweg und Sensorprüfung werden empfohlen.", }, { "titel": "Reparatur — Wartungsbetrieb", "modus": "reparatur_demo", "betriebsart": "wartung", "betroffen": "Instandhaltung, Reparaturhistorie", "auswirkung": "Reparaturstatus sichtbar, Wiederanlauf bleibt freigabepflichtig.", }, { "titel": "Prüfprotokoll erforderlich", "modus": "pruefprotokoll_erforderlich", "betriebsart": "wartung", "betroffen": "Prüfprotokolle, Wiederanlauf", "auswirkung": "Elektro-, Mechanik- oder Sensorprüfung und Meisterfreigabe erforderlich.", }, { "titel": "Wiederanlauf — Anlage Grün", "modus": "normal", "betriebsart": "automatik", "betroffen": "Gesamte Linie", "auswirkung": "Demo endet im sauberen Grünzustand, Historie bleibt dokumentiert.", }, ] def schreibe_json_datei(pfad, daten): """Schreibt JSON atomar, damit das Dashboard nie halbe Dateien liest.""" temp_pfad = f"{pfad}.tmp" with JSON_SCHREIB_LOCK: with open(temp_pfad, mode="w", encoding="utf-8") as datei: json.dump(daten, datei, indent=2, ensure_ascii=False) os.replace(temp_pfad, pfad) DASHBOARD_PFAD = os.path.join(SCRIPT_ORDNER, DASHBOARD_DATEI) STYLE_PFAD = os.path.join(SCRIPT_ORDNER, STYLE_DATEI) DATEN_PFAD = os.path.join(SCRIPT_ORDNER, DATEN_DATEI) HISTORIE_PFAD = os.path.join(SCRIPT_ORDNER, HISTORIE_DATEI) ALARM_HISTORIE_PFAD = os.path.join(SCRIPT_ORDNER, ALARM_HISTORIE_DATEI) QUITTIERUNGEN_PFAD = os.path.join(SCRIPT_ORDNER, QUITTIERUNGEN_DATEI) REPARATUR_HISTORIE_PFAD = os.path.join(SCRIPT_ORDNER, REPARATUR_HISTORIE_DATEI) SETTINGS_PFAD = os.path.join(SCRIPT_ORDNER, SETTINGS_DATEI) SIMULATION_STATE_PFAD = os.path.join(SCRIPT_ORDNER, SIMULATION_STATE_DATEI) ACTIVE_ALARM_STATE_PFAD = os.path.join(SCRIPT_ORDNER, ACTIVE_ALARM_STATE_DATEI) SCHICHTUEBERGABE_PFAD = os.path.join(SCRIPT_ORDNER, SCHICHTUEBERGABE_DATEI) TAGESBERICHT_PFAD = os.path.join(SCRIPT_ORDNER, TAGESBERICHT_DATEI) DASHBOARD_URL = f"http://localhost:{SERVER_PORT}/{DASHBOARD_DATEI}" HISTORIE_FELDNAMEN = [ "Zeitstempel", "Motorname", "Gesamtstatus", "Stromaufnahme", "Temperatur", "Vibration", "Drehzahl", "DurchschnittAbweichung", ] ALARM_HISTORIE_FELDNAMEN = [ "Zeitstempel", "Motorname", "Bereich", "Status", "Meldungstyp", "Risikowert", "Meldung", "Quittiert", ] REPARATUR_HISTORIE_FELDNAMEN = [ "Zeitstempel", "Motorname", "Bereich", "Fehlerbild", "Durchgefuehrte_Massnahme", "Ergebnis", "Kommentar", "Mitarbeiter", "Aktueller_Status", "Risikowert", "Schicht", "Fehlerart", "Mechanisches_Fehlerbild", "Elektrisches_Fehlerbild", "Mechanische_Massnahme", "Elektrische_Massnahme", "Teile_gewechselt", ] STANDARD_SETTINGS = { "abweichung_gelb_prozent": 10, "abweichung_rot_prozent": 25, "temperatur_gelb_c": 70, "temperatur_rot_c": 80, "vibration_extrem_mm_s": 20, "strom_extrem_faktor": 2.0, "strompreis_euro_kwh": 0.25, "aktualisierung_sekunden": 2, } def lade_settings(): """Laedt Grenzwerte aus settings.json oder erstellt Standardwerte.""" if not os.path.exists(SETTINGS_PFAD): with open(SETTINGS_PFAD, mode="w", encoding="utf-8") as datei: json.dump(STANDARD_SETTINGS, datei, indent=2, ensure_ascii=False) return dict(STANDARD_SETTINGS) try: with open(SETTINGS_PFAD, mode="r", encoding="utf-8") as datei: geladene_settings = json.load(datei) except json.JSONDecodeError: geladene_settings = {} settings = dict(STANDARD_SETTINGS) settings.update(geladene_settings) with open(SETTINGS_PFAD, mode="w", encoding="utf-8") as datei: json.dump(settings, datei, indent=2, ensure_ascii=False) return settings def lade_simulation_state(): """Laedt den aktuellen Demo-Simulationsmodus.""" standard_state = { "anlagenmodus": "standard", "betriebsart": "automatik", "praesentation_aktiv": False, "praesentation_schritt": 0, } if not os.path.exists(SIMULATION_STATE_PFAD): schreibe_json_datei(SIMULATION_STATE_PFAD, standard_state) return standard_state try: with open(SIMULATION_STATE_PFAD, mode="r", encoding="utf-8") as datei: state = json.load(datei) except json.JSONDecodeError: state = standard_state if state.get("anlagenmodus") not in SIMULATIONS_MODI: state["anlagenmodus"] = "standard" if state.get("betriebsart") not in BETRIEBSARTEN: state["betriebsart"] = "automatik" state["praesentation_aktiv"] = bool(state.get("praesentation_aktiv", False)) try: state["praesentation_schritt"] = int(state.get("praesentation_schritt", 0)) except (TypeError, ValueError): state["praesentation_schritt"] = 0 state["praesentation_schritt"] = max(0, min(state["praesentation_schritt"], len(PRAESENTATIONS_SCHRITTE) - 1)) return state def speichere_simulation_state(modus=None, betriebsart=None): """Speichert den Demo-Simulationsmodus.""" state = lade_simulation_state() if modus is not None: if modus not in SIMULATIONS_MODI: modus = "standard" state["anlagenmodus"] = modus state["praesentation_aktiv"] = False state["praesentation_schritt"] = 0 if betriebsart is not None: if betriebsart not in BETRIEBSARTEN: betriebsart = "automatik" state["betriebsart"] = betriebsart state["praesentation_aktiv"] = False state["praesentation_schritt"] = 0 schreibe_json_datei(SIMULATION_STATE_PFAD, state) return state def setze_praesentation(aktion): """Steuert die Demo-Story fuer Praesentationen.""" state = lade_simulation_state() schritt = int(state.get("praesentation_schritt", 0)) if aktion == "start": schritt = 0 state["praesentation_aktiv"] = True elif aktion == "next": state["praesentation_aktiv"] = True schritt = min(schritt + 1, len(PRAESENTATIONS_SCHRITTE) - 1) elif aktion == "prev": state["praesentation_aktiv"] = True schritt = max(schritt - 1, 0) elif aktion == "stop": state["praesentation_aktiv"] = False state["praesentation_schritt"] = schritt schreibe_json_datei(SIMULATION_STATE_PFAD, state) return state, False schritt_info = PRAESENTATIONS_SCHRITTE[schritt] state.update({ "praesentation_schritt": schritt, "anlagenmodus": schritt_info["modus"], "betriebsart": schritt_info["betriebsart"], }) schreibe_json_datei(SIMULATION_STATE_PFAD, state) return state, bool(schritt_info.get("auto_quittieren")) def quittiere_aktive_alarme_aus_dashboard(): """Quittiert aktive Alarme fuer die Praesentationsstory automatisch.""" daten = lese_vorherige_dashboard_daten() quittierungen = lese_quittierungen() zeitstempel = datetime.now().strftime("%Y-%m-%d %H:%M:%S") for alarm in daten.get("aktive_alarme", []): quittierungen[alarm_schluessel(alarm)] = { "status": alarm.get("status"), "zeitstempel": zeitstempel, } speichere_quittierungen(quittierungen) def lade_active_alarm_state(): """Laedt Startzeiten aktiver Alarme fuer Timer und Eskalation.""" if not os.path.exists(ACTIVE_ALARM_STATE_PFAD): schreibe_json_datei(ACTIVE_ALARM_STATE_PFAD, {}) return {} try: with open(ACTIVE_ALARM_STATE_PFAD, mode="r", encoding="utf-8") as datei: return json.load(datei) except json.JSONDecodeError: return {} def speichere_active_alarm_state(active_alarm_state): """Speichert Startzeiten aktiver Alarme.""" schreibe_json_datei(ACTIVE_ALARM_STATE_PFAD, active_alarm_state) def reset_active_alarm_state(): """Setzt die Alarm-Timer zurueck.""" speichere_active_alarm_state({}) def erzeuge_sps_daten(simulation_state): modus = simulation_state.get("anlagenmodus", "normal") sps_online = modus != "sicherheit_not_aus" cpu_last = round(random.uniform(35, 55), 1) if modus in ("cnc_stoerung", "kritisch") else round(random.uniform(18, 28), 1) return { "online": sps_online, "typ": "Siemens S7-1500 (Demo)", "betriebsart": simulation_state.get("betriebsart", "automatik"), "zykluszeit_ms": round(random.uniform(3.8, 4.2), 1), "cpu_last_prozent": cpu_last, "aktive_fehler": 0 if modus == "normal" else (2 if modus in ("kritisch", "cnc_stoerung") else 1), "aktive_warnungen": 0 if modus == "normal" else 1, "kommunikation_ok": sps_online, "letzte_quittierung": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "profinet_ok": sps_online, "hinweis": "Demo-SPS — keine echte Siemens-Verbindung", } def erzeuge_sicherheit_daten(simulation_state): modus = simulation_state.get("anlagenmodus", "normal") not_aus = modus == "sicherheit_not_aus" return { "not_aus_aktiv": not_aus, "schutztuer_offen": not_aus, "lichtgitter_unterbrochen": False, "sicherheitsrelais_ok": not not_aus, "sto_aktiv": not_aus, "quittierung_noetig": not_aus, "status": "Rot" if not_aus else "Gruen", "meldung": "Not-Aus aktiv — alle Antriebe gesperrt" if not_aus else "Sicherheitskreis OK", } def erzeuge_medien_daten(simulation_state): modus = simulation_state.get("anlagenmodus", "normal") luftdruck_niedrig = modus == "medien_luftdruck_niedrig" druck_ist = round(random.uniform(2.1, 2.8), 1) if luftdruck_niedrig else round(random.uniform(4.2, 4.8), 1) return { "luftdruck_soll_bar": 5.0, "luftdruck_ist_bar": druck_ist, "hydraulikdruck_bar": round(random.uniform(145, 155), 1), "kuehlschmierstoff_stand_prozent": random.randint(60, 90), "filterzustand": "Warnung" if modus in ("reinigung_stoerung", "reinigung_druckverlust") else "OK", "leckage_erkannt": False, "mediumtemperatur_c": round(random.uniform(18, 24), 1), "status": "Gelb" if luftdruck_niedrig else "Gruen", "meldung": "Luftdruck unter Sollwert — Pneumatikzylinder prüfen" if luftdruck_niedrig else "Medienversorgung OK", } def erzeuge_achsen_daten(simulation_state): modus = simulation_state.get("anlagenmodus", "normal") schwer = modus == "achse_schwergaengig" achsen = [] for ax_id, ax_name, pos_soll, nenn_strom in [ ("CNC_X", "X-Achse", 125.4, 8.2), ("CNC_Y", "Y-Achse", 87.2, 6.8), ("CNC_Z", "Z-Achse", 42.1, 5.5), ("CNC_U", "U-Achse", 0.0, 4.2), ]: faktor = random.uniform(1.18, 1.35) if (schwer and ax_id == "CNC_X") else random.uniform(0.95, 1.08) strom = round(nenn_strom * faktor, 1) temp = round(random.uniform(42, 52) + (8 if (schwer and ax_id == "CNC_X") else 0), 1) pos_ist = round(pos_soll + random.uniform(-0.02, 0.02), 2) risiko = 45 if (schwer and ax_id == "CNC_X") else random.randint(5, 18) achsen.append({ "id": ax_id, "name": ax_name, "station": "cnc_station", "position_soll_mm": pos_soll, "position_ist_mm": pos_ist, "positionsabweichung_mm": round(abs(pos_ist - pos_soll), 3), "motorstrom_a": strom, "temperatur_c": temp, "referenziert": True, "endlage_aktiv": False, "status": "Warnung" if risiko >= 40 else "OK", "risikowert": risiko, "ki_hinweis": f"{ax_name}: Strom erhöht, mögliche Schwergängigkeit" if (schwer and ax_id == "CNC_X") else f"{ax_name} OK", }) return achsen def erzeuge_reinigung_detail(simulation_state): modus = simulation_state.get("anlagenmodus", "normal") druckverlust = modus in ("reinigung_stoerung", "reinigung_druckverlust") druck_ist = round(random.uniform(55, 72), 1) if druckverlust else round(random.uniform(76, 84), 1) durchfluss = round(random.uniform(18, 24), 1) if druckverlust else round(random.uniform(32, 40), 1) return { "pumpe_an": True, "druck_soll_bar": 80.0, "druck_ist_bar": druck_ist, "durchfluss_l_min": durchfluss, "mediumtemperatur_c": round(random.uniform(55, 62), 1), "filterzustand": "Warnung" if druckverlust else "OK", "status": "Warnung" if druckverlust else "OK", "meldung": f"Reinigungsdruck {druck_ist} bar (Soll: 80 bar) — Filter oder Pumpe prüfen" if druckverlust else "Reinigungsprozess OK", } SETTINGS = lade_settings() AKTUALISIERUNG_SEKUNDEN = SETTINGS["aktualisierung_sekunden"] STROMPREIS_EURO_KWH = SETTINGS["strompreis_euro_kwh"] MAX_HISTORIE_ZEILEN = 50000 FEHLERCODE_KATALOG = [ { "fehlercode": "FB1-ERR-001", "meldung": "Motor startet nicht trotz Startsignal", "kategorie": "SPS I/O / Antrieb", "ausloeser": ["Q0.1 FU Start aktiv", "Drehzahl = 0", "Schrittkette haengt bei Motor starten"], "moegliche_ursachen": [ "Motorschutz ausgeloest", "Frequenzumrichter nicht bereit", "FU-Stoerung aktiv", "Drehzahlsensor defekt", "Motor mechanisch blockiert", "Ausgangssignal kommt nicht am FU an", ], "pruefschritte": [ "I0.1 Motorschutz OK pruefen", "I0.7 FU-Stoerung pruefen", "Q0.1 FU Start pruefen", "I0.5 Drehzahlimpuls pruefen", "FU-Fehlerspeicher auslesen", "Mechanik auf Blockade pruefen", ], "massnahmen": [ "FU-Fehlerspeicher auslesen", "Motorschutz pruefen", "Drehzahlsensor pruefen", "Mechanik freigaengig machen", "Startsignal an SPS-Ausgang und FU-Eingang vergleichen", ], "ersatzteile": [ "Drehzahlsensor FB1-SEN-RPM-IND", "Frequenzumrichter FB1-FU-5500", "Motor FB1-MOT-5500", "Sensorleitung / Klemmenmaterial", ], "gewerk": "Mechanik + Elektrik", "prioritaet": "Hoch", "kurzempfehlung": "FU, Motorschutz, Drehzahlsensor und Mechanik pruefen", "sps_adressen": ["Q0.1", "I0.1", "I0.5", "I0.7", "IW70"], }, { "fehlercode": "FB1-ERR-002", "meldung": "Vibration erhoeht", "kategorie": "Mechanik / Sensorik", "ausloeser": ["Vibration Gelb oder Rot", "Trend steigend"], "moegliche_ursachen": [ "Lagerproblem", "Unwucht", "lose Befestigung", "Kupplung falsch ausgerichtet", "Vibrationssensor lose", "Sensorfehler", ], "pruefschritte": [ "Lagergeraeusch pruefen", "Motorbefestigung pruefen", "Kupplung/Ausrichtung pruefen", "Vibrationssensorbefestigung pruefen", "Vergleichsmessung durchfuehren", "Reparaturhistorie pruefen: wurde Lager bereits gewechselt?", ], "massnahmen": [ "Lager pruefen", "Befestigung nachziehen", "Kupplung ausrichten", "Sensorbefestigung pruefen", "Nicht automatisch erneut Lager tauschen, falls bereits gewechselt", ], "ersatzteile": ["Lager FB1-LAG-6205-2RS", "Kupplung FB1-KUP-EL-01", "Vibrationssensor FB1-SEN-VIB-020"], "gewerk": "Mechanik", "prioritaet": "Mittel bis Hoch", "kurzempfehlung": "Lager, Kupplung, Befestigung und Sensorbefestigung pruefen", "sps_adressen": ["IW68"], }, { "fehlercode": "FB1-ERR-003", "meldung": "Temperatur unplausibel / Temperatursensorfehler", "kategorie": "Elektrik / Sensorik", "ausloeser": ["Temperatur <= -20 °C", "Temperatur springt stark", "Temperaturwert unplausibel"], "moegliche_ursachen": [ "PT100-Leitung unterbrochen", "Sensor nicht angeschlossen", "Wackelkontakt", "falsche Skalierung", "Analogeingang fehlerhaft", "Klemme lose", ], "pruefschritte": [ "Sensorleitung pruefen", "Klemme pruefen", "Widerstand / Signal pruefen", "Analogeingang pruefen", "Skalierung pruefen", "Sensor gegenpruefen", ], "massnahmen": ["Klemme nachziehen", "Sensorleitung reparieren", "Temperatursensor tauschen", "Skalierung korrigieren"], "ersatzteile": ["Temperatursensor FB1-SEN-T-PT100", "Klemmenmaterial", "Sensorleitung"], "gewerk": "Elektrik", "prioritaet": "Mittel bis Hoch", "kurzempfehlung": "PT100, Leitung, Klemme, Analogeingang und Skalierung pruefen", "sps_adressen": ["IW66"], }, { "fehlercode": "FB1-ERR-004", "meldung": "FU-Stoerung aktiv", "kategorie": "Frequenzumrichter / Antrieb", "ausloeser": ["I0.7 FU-Stoerung = 1", "SPS-I/O-Diagnose meldet FU-Stoerung"], "moegliche_ursachen": [ "Ueberstrom", "Ueberlast", "Parameterfehler", "Kommunikationsfehler", "Motor blockiert", "Rampe falsch eingestellt", "Netz-/Zwischenkreisproblem", ], "pruefschritte": [ "FU-Fehlerspeicher pruefen", "Ausgangsstrom pruefen", "Motorlast pruefen", "Parameter pruefen", "Sollwert/Istwert vergleichen", "Reset nur nach Ursachenpruefung", ], "massnahmen": ["Ursache im FU beseitigen", "Mechanik pruefen", "Parameter pruefen", "ggf. FU / Motor pruefen"], "ersatzteile": ["Frequenzumrichter FB1-FU-5500", "Motor FB1-MOT-5500"], "gewerk": "Elektrik + Mechanik", "prioritaet": "Hoch", "kurzempfehlung": "FU-Fehlerspeicher, Motorlast und Parameter pruefen", "sps_adressen": ["I0.7", "Q0.1", "QW64"], }, { "fehlercode": "FB1-ERR-005", "meldung": "Sensor S1/S2 Signal fehlt", "kategorie": "Sensorik / SPS-Eingang", "ausloeser": ["S1 oder S2 Status Fehler", "I0.2 oder I0.3 = 0", "Schrittkette wartet auf S1 oder S2"], "moegliche_ursachen": ["Sensor verschmutzt", "Sensor falsch ausgerichtet", "Kabelbruch", "SPS-Eingang defekt", "Materialstau", "kein Material vorhanden"], "pruefschritte": ["Sensor reinigen", "Sensorposition pruefen", "Signal am Sensor pruefen", "Signal an Klemme pruefen", "SPS-Eingang pruefen", "Materialfluss pruefen"], "massnahmen": ["Sensor reinigen", "Sensor einstellen", "Leitung pruefen", "Sensor tauschen", "Materialstau entfernen"], "ersatzteile": ["Sensor S1 FB1-SEN-IN-001", "Sensor S2 FB1-SEN-OUT-001", "Sensorleitung / Klemmenmaterial"], "gewerk": "Elektrik / Produktion", "prioritaet": "Mittel", "kurzempfehlung": "Sensor reinigen, ausrichten und Signalweg bis SPS pruefen", "sps_adressen": ["I0.2", "I0.3"], }, { "fehlercode": "FB1-ERR-006", "meldung": "Motorschutz ausgeloest", "kategorie": "Elektrik / Mechanik", "ausloeser": ["I0.1 Motorschutz OK = 0"], "moegliche_ursachen": ["Ueberlast", "Motorstrom zu hoch", "mechanische Schwergaengigkeit", "Kurzschluss / Wicklungsproblem", "Motorschutz falsch eingestellt"], "pruefschritte": ["Motorstrom messen", "Mechanik auf Schwergaengigkeit pruefen", "Motorschutz pruefen", "Motorwicklung pruefen", "Stromaufnahme im Dashboard vergleichen"], "massnahmen": ["Ursache fuer Ueberlast beseitigen", "Motorschutz erst nach Pruefung zuruecksetzen", "Mechanik pruefen", "Motor elektrisch pruefen"], "ersatzteile": ["Motor FB1-MOT-5500", "Klemmenmaterial", "Motorschutz / Schutzgeraet Demo"], "gewerk": "Mechanik + Elektrik", "prioritaet": "Hoch", "kurzempfehlung": "Motorschutz, Motorstrom und Mechanik pruefen", "sps_adressen": ["I0.1", "IW64"], }, { "fehlercode": "FB1-ERR-007", "meldung": "Bandlaufueberwachung meldet Fehler", "kategorie": "Mechanik / Sensorik", "ausloeser": ["Sensor S3 Bandlaufueberwachung Fehler", "I0.4 = 0"], "moegliche_ursachen": ["Band laeuft schief", "Sensor falsch eingestellt", "Bandfuehrung verschmutzt", "mechanische Fuehrung defekt", "Sensorleitung fehlerhaft"], "pruefschritte": ["Bandlauf pruefen", "Bandfuehrung pruefen", "Sensorabstand pruefen", "Sensorleitung pruefen", "Materialstau ausschliessen"], "massnahmen": ["Bandlauf einstellen", "Sensor neu ausrichten", "Fuehrung reinigen", "Sensor/Kabel pruefen"], "ersatzteile": ["Bandschieflaufsensor FB1-SEN-BAND-001", "Klemmenmaterial", "Sensorleitung"], "gewerk": "Mechanik + Elektrik", "prioritaet": "Mittel", "kurzempfehlung": "Bandlauf, Sensorabstand und Sensorleitung pruefen", "sps_adressen": ["I0.4"], }, { "fehlercode": "FB1-ERR-008", "meldung": "Materialstau / Materialfluss gestoert", "kategorie": "Produktion / Sensorik", "ausloeser": ["S1 erkennt Material", "S2 erkennt keinen Ausgang", "Schrittkette bleibt bei Foerdern"], "moegliche_ursachen": ["Materialstau zwischen S1 und S2", "Bandbereich blockiert", "Sensor S2 Signal fehlt", "Pruefstation nimmt Teil nicht ab"], "pruefschritte": ["Bandbereich zwischen S1 und S2 pruefen", "Materialfluss pruefen", "Sensor S2 I0.3 pruefen", "Pruefstation und Ausgang kontrollieren"], "massnahmen": ["Stau entfernen", "Sensor S2 pruefen", "Bandlauf pruefen", "Materialzufuhr reduzieren"], "ersatzteile": ["Sensor S2 FB1-SEN-OUT-001", "Sensorleitung / Klemmenmaterial"], "gewerk": "Produktion + Elektrik", "prioritaet": "Mittel", "kurzempfehlung": "Bandbereich, S2 und Materialfluss pruefen", "sps_adressen": ["I0.2", "I0.3"], }, ] MOTOREN = [ { "name": "Motor Foerderband 1", "bereich": "Linie 1 / Foerdertechnik", "zustand": "normal", "produktionsrelevanz": "Hoch - Ausfall stoppt ganze Linie", "betriebsstunden": 4820, "wartungsintervall_stunden": 5000, "nennleistung_kw": 5.5, "wirtschaft": { "stillstandskosten_pro_stunde": 2500, "ausfalldauer_ohne_fruehwarnung": 4, "geplante_wartungskosten": 800, }, "maschinenakte": { "anlagenteil": "Foerdertechnik Linie 1", "leistung": "5,5 kW", "baujahr": 2018, "letzte_wartung": "2026-04-12", "naechste_geplante_wartung": "2026-05-20", "inventarnummer": "FB1-M-001", }, "normalwerte": { "Stromaufnahme": 12.0, "Temperatur": 65.0, "Vibration": 3.0, "Drehzahl": 1450.0, }, }, { "name": "Motor Pumpe 2", "bereich": "Pumpenstation", "zustand": "auffaellig", "produktionsrelevanz": "Mittel - Ausfall reduziert Leistung", "betriebsstunden": 3100, "wartungsintervall_stunden": 4000, "nennleistung_kw": 3.0, "wirtschaft": { "stillstandskosten_pro_stunde": 1800, "ausfalldauer_ohne_fruehwarnung": 3, "geplante_wartungskosten": 600, }, "maschinenakte": { "anlagenteil": "Pumpenstation Kuehlkreislauf", "leistung": "3,0 kW", "baujahr": 2020, "letzte_wartung": "2026-03-28", "naechste_geplante_wartung": "2026-05-18", "inventarnummer": "PU2-M-014", }, "normalwerte": { "Stromaufnahme": 8.5, "Temperatur": 60.0, "Vibration": 2.5, "Drehzahl": 2900.0, }, }, { "name": "Motor Luefter 3", "bereich": "Lueftungsanlage", "zustand": "kritisch", "produktionsrelevanz": "Mittel - Ausfall reduziert Leistung", "betriebsstunden": 5960, "wartungsintervall_stunden": 6000, "nennleistung_kw": 2.2, "wirtschaft": { "stillstandskosten_pro_stunde": 1200, "ausfalldauer_ohne_fruehwarnung": 5, "geplante_wartungskosten": 700, }, "maschinenakte": { "anlagenteil": "Lueftungsanlage Halle 1", "leistung": "2,2 kW", "baujahr": 2017, "letzte_wartung": "2026-04-02", "naechste_geplante_wartung": "2026-05-15", "inventarnummer": "LU3-M-009", }, "normalwerte": { "Stromaufnahme": 5.0, "Temperatur": 58.0, "Vibration": 1.8, "Drehzahl": 1500.0, }, }, ] EINHEITEN = { "Stromaufnahme": "A", "Temperatur": "°C", "Vibration": "mm/s", "Drehzahl": "U/min", } SIMULATIONS_BEREICHE = { "normal": { "Stromaufnahme": (0.96, 1.06), "Temperatur": (0.92, 1.05), "Vibration": (0.90, 1.08), "Drehzahl": (0.98, 1.02), }, "auffaellig": { "Stromaufnahme": (1.12, 1.22), "Temperatur": (1.08, 1.18), "Vibration": (1.18, 1.35), "Drehzahl": (0.94, 0.98), }, "kritisch": { "Stromaufnahme": (1.28, 1.45), "Temperatur": (1.25, 1.45), "Vibration": (1.50, 1.90), "Drehzahl": (0.88, 0.94), }, } SIMULATIONS_BEREICHE_WARNBETRIEB = { "Stromaufnahme": (1.12, 1.20), "Temperatur": (1.04, 1.08), "Vibration": (1.12, 1.22), "Drehzahl": (0.96, 0.98), } WARTUNGSFENSTER = [ { "name": "Samstag Spaetschicht", "in_tagen": 2, "dauer_stunden": 6, "beschreibung": "Produktionslinie steht, Wartung gut moeglich", }, { "name": "Sonntag Fruehschicht", "in_tagen": 3, "dauer_stunden": 8, "beschreibung": "Geplanter Anlagenstillstand", }, { "name": "Mittwoch Nachtschicht", "in_tagen": 7, "dauer_stunden": 4, "beschreibung": "Reduzierte Produktion", }, ] PRODUKTIONS_GRUNDWERTE = { "linie": "Linie 1 / Foerdertechnik", "sollleistung_pro_stunde": 120, "wert_pro_teil": 12, "schichtdauer_stunden": 8, } PRODUKTIONS_AUSWIRKUNGEN = { "Motor Foerderband 1": { "relevanz": "Hoch", "auswirkung": "Ausfall stoppt Linie 1", "teileverlust_pro_stunde": 120, "empfehlung": "Motor schnell pruefen, da direkter Einfluss auf Produktion", }, "Motor Pumpe 2": { "relevanz": "Mittel", "auswirkung": "Leistung reduziert, Kuehlung beeintraechtigt", "teileverlust_pro_stunde": 40, "empfehlung": "Temperatur und Pumpenlauf beobachten", }, "Motor Luefter 3": { "relevanz": "Mittel", "auswirkung": "Temperatur-/Luftproblem, Leistung reduziert", "teileverlust_pro_stunde": 30, "empfehlung": "Luefterzustand pruefen, da Umgebungseinfluss moeglich", }, } SPS_ZUORDNUNG_FOERDERBAND_1 = { "Motorfreigabe": "Q0.0", "FU Start": "Q0.1", "FU Stoerung": "I0.7", "Motorschutz OK": "I0.1", "Stromaufnahme": "IW64", "Temperatur": "IW66", "Vibration": "IW68", "Drehzahl": "IW70", "Drehzahlimpuls": "I0.5", "Sensor S1 Signal": "I0.2", "Sensor S2 Signal": "I0.3", "Sensor S3 Signal": "I0.4", } VERBAUTE_BAUTEILE = { "Motor Foerderband 1": [ {"gruppe": "Motor", "bezeichnung": "Drehstrommotor Foerderband 1", "typ": "M3AA-132S", "signal": "-", "teilenummer": "FB1-MOT-5500", "ersatzteilhinweis": "Motor nur bei Wicklungsschaden oder Lagerschaden tauschen"}, {"gruppe": "Sensorik", "bezeichnung": "Stromsensor Antrieb Foerderband 1", "typ": "0-20 A Messwandler", "signal": "4-20 mA", "teilenummer": "FB1-SEN-I-020", "ersatzteilhinweis": "Bei unplausibler Stromaufnahme Sensorleitung und Messwandler pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Motortemperaturfuehler", "typ": "PT100 / Temperaturfuehler", "signal": "analog", "teilenummer": "FB1-SEN-T-PT100", "ersatzteilhinweis": "Bei -20 °C oder unrealistischen Spruengen Sensor und Leitung pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Schwingungssensor Motorlager", "typ": "Vibrationssensor 0-20 mm/s", "signal": "4-20 mA", "teilenummer": "FB1-SEN-VIB-020", "ersatzteilhinweis": "Bei 0 mm/s ueber mehrere Messungen Sensorbefestigung und Kabel pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Drehzahlsensor Foerderband", "typ": "induktiver Naeherungssensor", "signal": "digital / Impuls", "teilenummer": "FB1-SEN-RPM-IND", "ersatzteilhinweis": "Bei Drehzahl 0 trotz laufendem Motor Sensorabstand und Leitung pruefen"}, {"gruppe": "Antrieb", "bezeichnung": "Frequenzumrichter Foerderband 1", "typ": "FU 400 V / 5,5 kW", "signal": "Feldbus / analog", "teilenummer": "FB1-FU-5500", "ersatzteilhinweis": "Bei Drehzahlabweichung Parameter, Sollwert und Fehlerspeicher pruefen"}, {"gruppe": "Mechanik", "bezeichnung": "Motorlager Antriebsseite", "typ": "Rillenkugellager 6205 2RS", "signal": "-", "teilenummer": "FB1-LAG-6205-2RS", "ersatzteilhinweis": "Bei Vibration und erhoehter Temperatur Lager pruefen, aber Historie beachten"}, {"gruppe": "Mechanik", "bezeichnung": "Kupplung Motor-Foerderband", "typ": "elastische Kupplung", "signal": "-", "teilenummer": "FB1-KUP-EL-01", "ersatzteilhinweis": "Bei Vibration trotz neuem Lager Ausrichtung/Kupplung pruefen"}, ], "Motor Pumpe 2": [ {"gruppe": "Motor", "bezeichnung": "Drehstrommotor Pumpe 2", "typ": "PU2-MOT-3000", "signal": "-", "teilenummer": "PU2-MOT-3000", "ersatzteilhinweis": "Motor nur nach elektrischer und mechanischer Pruefung tauschen"}, {"gruppe": "Sensorik", "bezeichnung": "Stromsensor Pumpe 2", "typ": "0-15 A Messwandler", "signal": "4-20 mA", "teilenummer": "PU2-SEN-I-015", "ersatzteilhinweis": "Messwandler und Analogeingang gegenpruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Temperatursensor Pumpe 2", "typ": "PT100", "signal": "analog", "teilenummer": "PU2-SEN-T-PT100", "ersatzteilhinweis": "Sensor, Leitung und Eingangskarte pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Vibrationssensor Pumpe 2", "typ": "Vibrationssensor 0-20 mm/s", "signal": "4-20 mA", "teilenummer": "PU2-SEN-VIB-020", "ersatzteilhinweis": "Sensorbefestigung und Kabel pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Drehzahlsensor Pumpe 2", "typ": "induktiver Naeherungssensor", "signal": "digital / Impuls", "teilenummer": "PU2-SEN-RPM-IND", "ersatzteilhinweis": "Sensorabstand und Impulsgeber pruefen"}, {"gruppe": "Antrieb", "bezeichnung": "Frequenzumrichter Pumpe 2", "typ": "FU 400 V / 3,0 kW", "signal": "Feldbus / analog", "teilenummer": "PU2-FU-3000", "ersatzteilhinweis": "FU-Fehlerspeicher und Parameter pruefen"}, {"gruppe": "Mechanik", "bezeichnung": "Pumpenlager", "typ": "Rillenkugellager 6204 2RS", "signal": "-", "teilenummer": "PU2-LAG-6204-2RS", "ersatzteilhinweis": "Bei Vibration und Laufgeraeusch pruefen"}, {"gruppe": "Mechanik", "bezeichnung": "Gleitringdichtung", "typ": "Gleitringdichtung", "signal": "-", "teilenummer": "PU2-DICHT-GRD-01", "ersatzteilhinweis": "Bei Leckage oder Trockenlauf pruefen"}, ], "Motor Luefter 3": [ {"gruppe": "Motor", "bezeichnung": "Drehstrommotor Luefter 3", "typ": "LU3-MOT-2200", "signal": "-", "teilenummer": "LU3-MOT-2200", "ersatzteilhinweis": "Motor nach Lager- und Wicklungspruefung tauschen"}, {"gruppe": "Sensorik", "bezeichnung": "Stromsensor Luefter 3", "typ": "0-10 A Messwandler", "signal": "4-20 mA", "teilenummer": "LU3-SEN-I-010", "ersatzteilhinweis": "Messsignal und Skalierung pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Temperatursensor Luefter 3", "typ": "PT100", "signal": "analog", "teilenummer": "LU3-SEN-T-PT100", "ersatzteilhinweis": "Temperatursensor und Leitung pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Vibrationssensor Luefter 3", "typ": "Vibrationssensor 0-20 mm/s", "signal": "4-20 mA", "teilenummer": "LU3-SEN-VIB-020", "ersatzteilhinweis": "Sensorbefestigung, Kabel und Eingang pruefen"}, {"gruppe": "Sensorik", "bezeichnung": "Drehzahlsensor Luefter 3", "typ": "induktiver Naeherungssensor", "signal": "digital / Impuls", "teilenummer": "LU3-SEN-RPM-IND", "ersatzteilhinweis": "Sensorabstand und Signal pruefen"}, {"gruppe": "Antrieb", "bezeichnung": "Frequenzumrichter Luefter 3", "typ": "FU 400 V / 2,2 kW", "signal": "Feldbus / analog", "teilenummer": "LU3-FU-2200", "ersatzteilhinweis": "Sollwert, Rampe und Fehlerspeicher pruefen"}, {"gruppe": "Mechanik", "bezeichnung": "Luefterlager", "typ": "Rillenkugellager 6203 2RS", "signal": "-", "teilenummer": "LU3-LAG-6203-2RS", "ersatzteilhinweis": "Bei Vibration Lager pruefen"}, {"gruppe": "Mechanik", "bezeichnung": "Keilriemen", "typ": "SPZ-01", "signal": "-", "teilenummer": "LU3-RIEMEN-SPZ-01", "ersatzteilhinweis": "Bei Drehzahlabweichung Riemenspannung pruefen"}, ], } ZUSATZSENSOREN_FOERDERBAND = [ { "id": "FB1-S1", "sensorname": "S1 Materialeingang", "bereich": "Linie 1 / Foerdertechnik", "zugehoeriger_motor": "Motor Foerderband 1", "signaltyp": "Digital", "teilenummer": "FB1-SEN-IN-001", "funktion": "Material am Eingang erkannt", "moegliche_fehler": ["Sensor verschmutzt", "Leitung unterbrochen", "Ausrichtung falsch"], "empfohlene_pruefung": ["Sensor reinigen", "Signal an SPS-Eingang pruefen", "Sensorposition kontrollieren"], }, { "id": "FB1-S2", "sensorname": "S2 Materialausgang", "bereich": "Linie 1 / Foerdertechnik", "zugehoeriger_motor": "Motor Foerderband 1", "signaltyp": "Digital", "teilenummer": "FB1-SEN-OUT-001", "funktion": "Material am Ausgang erkannt", "moegliche_fehler": ["Reflektor verschmutzt", "Kabelbruch", "SPS-Eingang ohne Signal"], "empfohlene_pruefung": ["Optik reinigen", "Schaltsignal beobachten", "Klemmen pruefen"], }, { "id": "FB1-S3", "sensorname": "S3 Bandlaufueberwachung", "bereich": "Linie 1 / Foerdertechnik", "zugehoeriger_motor": "Motor Foerderband 1", "signaltyp": "Digital", "teilenummer": "FB1-SEN-BAND-001", "funktion": "Bandlauf / Schieflauf ueberwachen", "moegliche_fehler": ["Band schief", "Sensor falsch eingestellt", "Signalstoerung"], "empfohlene_pruefung": ["Bandlauf pruefen", "Sensorabstand einstellen", "Mechanische Fuehrung kontrollieren"], }, ] def berechne_abweichung_prozent(normalwert, aktueller_wert): """Berechnet die prozentuale Abweichung vom Normalwert.""" if normalwert == 0: return 0.0 return ((aktueller_wert - normalwert) / normalwert) * 100 def bewerte_abweichung(abweichung_prozent): """Bewertet Stromaufnahme, Vibration und Drehzahl.""" betrag = abs(abweichung_prozent) if betrag >= SETTINGS["abweichung_rot_prozent"]: return "Rot" if betrag >= SETTINGS["abweichung_gelb_prozent"]: return "Gelb" return "Gruen" def bewerte_temperatur(aktueller_wert): """Bewertet Temperatur mit festen Grenzwerten.""" if aktueller_wert >= SETTINGS["temperatur_rot_c"]: return "Rot" if aktueller_wert >= SETTINGS["temperatur_gelb_c"]: return "Gelb" return "Gruen" def ermittle_gesamtstatus(status_liste): """Der schlechteste Einzelstatus bestimmt den Gesamtstatus.""" if "Rot" in status_liste: return "Rot" if "Gelb" in status_liste: return "Gelb" return "Gruen" def ermittle_warnmeldung(messwert_name, status, abweichung_prozent): """Erstellt technische Warnmeldungen fuer auffaellige Werte.""" if status == "Gruen": return "Keine Auffaelligkeit" if messwert_name == "Stromaufnahme": if abweichung_prozent > 0: return ( "Stromaufnahme erhoeht: moeglicher mechanischer Widerstand, " "Ueberlast oder schwergaengige Mechanik" ) return ( "Stromaufnahme niedriger als normal: moegliche Unterlast, " "Sensorfehler oder fehlende Last" ) if messwert_name == "Temperatur": if status == "Rot": return "Temperatur kritisch hoch: Motor zeitnah pruefen, Ueberhitzungsgefahr" return ( "Temperatur erhoeht: Kuehlung, Luefter, Umgebungstemperatur " "oder Belastung pruefen" ) if messwert_name == "Vibration": return ( "Erhoehte Vibration: moegliche Unwucht, Lagerproblem, " "lose Befestigung oder mechanischer Verschleiss" ) if messwert_name == "Drehzahl": return "Drehzahl abweichend: Antrieb, Last, Frequenzumrichter oder Regelung pruefen" return "Wert beobachten" def ermittle_kombinationsdiagnosen(ergebnisse): """Erkennt einfache Fehlerbilder aus mehreren auffaelligen Messwerten.""" werte = {ergebnis["messwert"]: ergebnis for ergebnis in ergebnisse} strom_hoch = werte["Stromaufnahme"]["abweichung_prozent"] >= SETTINGS["abweichung_gelb_prozent"] temperatur_hoch = werte["Temperatur"]["status"] in ["Gelb", "Rot"] vibration_hoch = werte["Vibration"]["abweichung_prozent"] >= SETTINGS["abweichung_gelb_prozent"] drehzahl_niedrig = werte["Drehzahl"]["abweichung_prozent"] <= -SETTINGS["abweichung_gelb_prozent"] drehzahl_normal = abs(werte["Drehzahl"]["abweichung_prozent"]) < SETTINGS["abweichung_gelb_prozent"] diagnosen = [] if strom_hoch and vibration_hoch: diagnosen.append( "Kombination: Stromaufnahme hoch und Vibration hoch - " "moegliches Lagerproblem oder mechanischer Widerstand" ) if strom_hoch and drehzahl_niedrig: diagnosen.append( "Kombination: Stromaufnahme hoch und Drehzahl niedrig - " "Ueberlast oder schwergaengige Mechanik" ) if temperatur_hoch and strom_hoch: diagnosen.append( "Kombination: Temperatur hoch und Stromaufnahme hoch - " "thermische Ueberlast" ) if vibration_hoch and drehzahl_normal: diagnosen.append( "Kombination: Vibration hoch und Drehzahl normal - " "Unwucht oder lose Befestigung" ) if temperatur_hoch and vibration_hoch: diagnosen.append( "Kombination: Temperatur hoch und Vibration hoch - " "Lager erwaermt sich, mechanischer Verschleiss moeglich" ) return diagnosen def berechne_risikowert( gesamtstatus, durchschnittsabweichung, langzeittrend, kombinationsdiagnosen, produktionsrelevanz, ): """Berechnet einen einfachen Demo-Risikowert von 0 bis 100.""" grundwerte = { "Gruen": 10, "Gelb": 45, "Rot": 75, } risiko = grundwerte[gesamtstatus] risiko += min(durchschnittsabweichung, 20) if langzeittrend == "Langzeittrend auffaellig": risiko += 10 elif langzeittrend == "Langzeittrend kritisch": risiko += 20 risiko += len(kombinationsdiagnosen) * 5 if produktionsrelevanz.startswith("Hoch"): risiko += 10 elif produktionsrelevanz.startswith("Mittel"): risiko += 5 return min(round(risiko), 100) def berechne_ausfallprognose(risikowert): """Leitet aus dem Risikowert eine grobe simulierte Ausfallprognose ab.""" if risikowert < 30: return ( "Kein kurzfristiger Ausfall zu erwarten (Simulation)", 31, ) if risikowert < 60: return ( "Auffaelligkeit vorhanden, Pruefung einplanen (Simulation)", 14, ) if risikowert < 80: return ( "Erhoehtes Ausfallrisiko, Pruefung zeitnah planen (Simulation)", 7, ) return ( "Kritisch: moeglicher Ausfall kurzfristig (Simulation)", 3, ) def erstelle_wartungsempfehlung(gesamtstatus, risikowert): """Erstellt eine einfache Empfehlung fuer die Instandhaltung.""" if risikowert >= 90: return "Sofortige Pruefung empfohlen. Austausch von Lager/Motor vorbereiten." if gesamtstatus == "Rot": return "Zeitnahe Wartung erforderlich. Pruefung innerhalb von 3 bis 7 Tagen empfohlen." if gesamtstatus == "Gelb": return "Wartung innerhalb der naechsten 7 bis 14 Tage einplanen. Werte weiter beobachten." return "Keine sofortige Massnahme erforderlich. Naechste Routinepruefung reicht aus." def waehle_wartungsfenster(geschaetzte_tage_bis_ausfall): """Waehlt ein Wartungsfenster, das vor der geschaetzten Risikofrist liegt.""" passende_fenster = [ fenster for fenster in WARTUNGSFENSTER if fenster["in_tagen"] <= geschaetzte_tage_bis_ausfall ] if not passende_fenster: return "Kein passendes Wartungsfenster vorhanden - ausserplanmaessige Wartung pruefen." fenster = sorted(passende_fenster, key=lambda eintrag: eintrag["in_tagen"])[0] return ( f"{fenster['name']} in {fenster['in_tagen']} Tagen, " f"{fenster['dauer_stunden']} Stunden - {fenster['beschreibung']}" ) def ermittle_empfohlene_massnahmen(ergebnisse): """Leitet konkrete Massnahmen aus auffaelligen Einzelwerten ab.""" massnahmen = [] auffaellige_werte = [ ergebnis for ergebnis in ergebnisse if ergebnis["status"] != "Gruen" ] for ergebnis in auffaellige_werte: messwert = ergebnis["messwert"] if messwert == "Stromaufnahme": massnahmen.extend( [ "Stromaufnahme am Umrichter pruefen", "Mechanische Last pruefen", "Foerderband/Pumpe auf Schwergaengigkeit pruefen", ] ) elif messwert == "Temperatur": massnahmen.extend( [ "Luefter/Kuehlung pruefen", "Umgebungstemperatur pruefen", "Motor auf Ueberlast pruefen", ] ) elif messwert == "Vibration": massnahmen.extend( [ "Lager pruefen", "Motorbefestigung pruefen", "Kupplung/Ausrichtung pruefen", "Unwucht pruefen", ] ) elif messwert == "Drehzahl": massnahmen.extend( [ "Frequenzumrichter pruefen", "Lastwechsel pruefen", "Regelung/Sollwert pruefen", ] ) if not massnahmen: return ["Routinekontrolle gemaess Wartungsplan fortsetzen"] # Doppelte Massnahmen entfernen, Reihenfolge bleibt erhalten. eindeutige_massnahmen = [] for massnahme in massnahmen: if massnahme not in eindeutige_massnahmen: eindeutige_massnahmen.append(massnahme) return eindeutige_massnahmen def ermittle_ersatzteile(ergebnisse): """Leitet Ersatzteile und Pruefbereiche aus auffaelligen Messwerten ab.""" empfehlungen = [] for ergebnis in ergebnisse: if ergebnis["status"] == "Gruen": continue messwert = ergebnis["messwert"] if messwert == "Stromaufnahme": empfehlungen.extend( [ "Motor pruefen", "Mechanische Baugruppe pruefen", "Frequenzumrichter pruefen", ] ) elif messwert == "Temperatur": empfehlungen.extend( [ "Luefter pruefen / Ersatzluefter bereitlegen", "Kuehleinheit pruefen", "Temperatursensor pruefen", ] ) elif messwert == "Vibration": empfehlungen.extend( [ "Lager pruefen / Ersatzlager bereitlegen", "Kupplung und Ausrichtung pruefen", "Befestigungsmaterial pruefen", ] ) elif messwert == "Drehzahl": empfehlungen.extend( [ "Frequenzumrichter pruefen", "Drehzahlsensor pruefen", "Regelung/Sollwert pruefen", ] ) if not empfehlungen: return ["Keine Ersatzteile erforderlich, Routinebestand ausreichend"] eindeutige_empfehlungen = [] for empfehlung in empfehlungen: if empfehlung not in eindeutige_empfehlungen: eindeutige_empfehlungen.append(empfehlung) return eindeutige_empfehlungen def berechne_wirtschaftliche_bewertung(motor): """Berechnet eine einfache Demo-Schaetzung fuer Kosten und Einsparpotenzial.""" wirtschaft = motor["wirtschaft"] ausfallkosten = ( wirtschaft["stillstandskosten_pro_stunde"] * wirtschaft["ausfalldauer_ohne_fruehwarnung"] ) einsparpotenzial = ausfallkosten - wirtschaft["geplante_wartungskosten"] return { "hinweis": "Demo-Schaetzung, keine garantierte Kostenrechnung", "stillstandskosten_pro_stunde": wirtschaft["stillstandskosten_pro_stunde"], "ausfalldauer_ohne_fruehwarnung": wirtschaft["ausfalldauer_ohne_fruehwarnung"], "moegliche_ausfallkosten": ausfallkosten, "geplante_wartungskosten": wirtschaft["geplante_wartungskosten"], "moegliches_einsparpotenzial": einsparpotenzial, } def ermittle_meldungstyp(gesamtstatus): """Unterscheidet zwischen keiner Meldung, Warnung und kritischer Stoerung.""" if gesamtstatus == "Rot": return "Kritische Stoerung" if gesamtstatus == "Gelb": return "Warnung" return "Keine Meldung" def ermittle_prioritaet(risikowert): """Leitet eine einfache Instandhaltungsprioritaet aus dem Risiko ab.""" if risikowert >= 80: return "Prioritaet 1 - Sofort pruefen" if risikowert >= 60: return "Prioritaet 2 - Zeitnah pruefen" if risikowert >= 30: return "Prioritaet 3 - Einplanen" return "Prioritaet 4 - Routine" def pruefe_plausibilitaet(motor, aktuelle_werte, historie): """Erkennt einfache unplausible Sensorwerte.""" warnungen = [] normalwerte = motor["normalwerte"] temperatur = aktuelle_werte["Temperatur"] drehzahl = aktuelle_werte["Drehzahl"] vibration = aktuelle_werte["Vibration"] strom = aktuelle_werte["Stromaufnahme"] normalstrom = normalwerte["Stromaufnahme"] if temperatur < -10: warnungen.append("Temperatur unter -10 °C: Sensorfehler moeglich") if temperatur > 130: warnungen.append("Temperatur ueber 130 °C: Messwert unplausibel / Sensor pruefen") if drehzahl == 0 and strom > normalstrom * (1 + SETTINGS["abweichung_gelb_prozent"] / 100): warnungen.append("Drehzahl 0 bei hoher Stromaufnahme: Motor blockiert oder Drehzahlsensor fehlerhaft") elif drehzahl == 0: warnungen.append("Drehzahl 0 bei normaler Stromaufnahme: Drehzahlsensor pruefen") if vibration > SETTINGS["vibration_extrem_mm_s"]: warnungen.append(f"Vibration ueber {SETTINGS['vibration_extrem_mm_s']} mm/s: Sensorfehler oder schwerer mechanischer Fehler") letzte_vibrationen = historie.get(f"{motor['name']}:Vibration", []) if vibration == 0 and len(letzte_vibrationen) >= 2 and all(wert == 0 for wert in letzte_vibrationen[-2:]): warnungen.append("Vibration ueber mehrere Messungen 0: Vibrationssensor pruefen") if strom == 0 and drehzahl > 0: warnungen.append("Stromaufnahme 0 bei Drehzahl groesser 0: Stromsensor pruefen") if strom > normalstrom * SETTINGS["strom_extrem_faktor"]: warnungen.append("Stromaufnahme ueber 200 % vom Normalwert: Messwert pruefen / Ueberlast moeglich") return warnungen def erstelle_elektronikfehler(kategorie, diagnose, ursachen, pruefungen, bauteile, prioritaet): """Erstellt einen strukturierten Elektronik-/Sensorikfehler.""" return { "kategorie": kategorie, "diagnose": diagnose, "moegliche_ursachen": ursachen, "empfohlene_pruefung": pruefungen, "betroffene_bauteile": bauteile, "prioritaet": prioritaet, } def diagnostiziere_elektronikfehler(motor, aktuelle_werte, historie): """Erkennt einfache elektrische, elektronische und sensorische Auffaelligkeiten.""" fehler = [] motorname = motor["name"] normalwerte = motor["normalwerte"] temperatur = aktuelle_werte["Temperatur"] vibration = aktuelle_werte["Vibration"] drehzahl = aktuelle_werte["Drehzahl"] strom = aktuelle_werte["Stromaufnahme"] letzte_temperaturen = historie.get(f"{motorname}:Temperatur", []) letzte_drehzahlen = historie.get(f"{motorname}:Drehzahl", []) letzte_stroeme = historie.get(f"{motorname}:Stromaufnahme", []) letzte_vibrationen = historie.get(f"{motorname}:Vibration", []) if temperatur <= -20: fehler.append(erstelle_elektronikfehler( "Temperatursensor", "Temperatursensor zeigt unrealistisch niedrigen Wert", ["Sensorleitung unterbrochen", "Sensor nicht angeschlossen", "falscher Eingang / falsche Skalierung", "PT100 / Analogsignal fehlerhaft"], ["Temperatursensor und Anschluss pruefen", "Leitungsbruch messen", "Analogeingang pruefen", "Sensor ggf. tauschen"], ["Temperatursensor", "Analogeingang / SPS-Modul", "Sensorleitung"], "Hoch", )) if letzte_temperaturen and abs(temperatur - letzte_temperaturen[-1]) > 30: fehler.append(erstelle_elektronikfehler( "Temperatursensor", "Temperatursignal springt unplausibel", ["Wackelkontakt", "lose Klemme", "EMV-Stoerung", "defekter Sensor"], ["Klemmen pruefen", "Kabel bewegen und Signal beobachten", "Schirmung pruefen", "Sensor tauschen, falls Fehler reproduzierbar"], ["Temperatursensor", "Sensorleitung", "Analogeingang / SPS-Modul"], "Mittel", )) if vibration == 0 and len(letzte_vibrationen) >= 2 and all(wert == 0 for wert in letzte_vibrationen[-2:]): fehler.append(erstelle_elektronikfehler( "Vibrationssensor", "Vibrationssensor liefert kein Signal", ["Sensor lose", "Sensor defekt", "Kabelbruch", "Versorgungsspannung fehlt", "Eingangskanal defekt"], ["Sensorbefestigung pruefen", "Versorgungsspannung messen", "Signal 4-20 mA pruefen", "Sensor ggf. tauschen"], ["Vibrationssensor", "Sensorleitung", "Analogeingang / SPS-Modul"], "Hoch", )) if vibration > SETTINGS["vibration_extrem_mm_s"]: fehler.append(erstelle_elektronikfehler( "Vibrationssensor", "Vibrationswert extrem hoch", ["schwerer mechanischer Fehler", "Sensor lose montiert", "falsche Skalierung", "Stoersignal"], ["Mechanik sofort pruefen", "Sensorbefestigung pruefen", "Skalierung im Programm pruefen", "Vergleichsmessung durchfuehren"], ["Vibrationssensor", "Lager", "Kupplung"], "Hoch", )) if drehzahl == 0 and strom > normalwerte["Stromaufnahme"] * (1 + SETTINGS["abweichung_gelb_prozent"] / 100): fehler.append(erstelle_elektronikfehler( "Drehzahlsensor", "Motor zieht Strom, aber Drehzahl wird nicht erkannt", ["Motor blockiert", "Drehzahlsensor defekt", "Sensorabstand falsch", "Impulsgeber verschmutzt", "Kabelbruch"], ["mechanische Blockade pruefen", "Drehzahlsensor pruefen", "Sensorabstand einstellen", "Impulsgeber reinigen", "Eingangssignal an SPS pruefen"], ["Drehzahlsensor", "Sensorleitung", "SPS-Eingang", "Mechanik"], "Hoch", )) elif drehzahl == 0: fehler.append(erstelle_elektronikfehler( "Drehzahlsensor", "Drehzahlsignal fehlt vermutlich", ["Drehzahlsensor defekt", "Leitung unterbrochen", "Eingangskanal defekt", "falscher Sollwert / Signal nicht freigegeben"], ["Sensor und Leitung pruefen", "SPS-Eingang pruefen", "Signalstatus im Programm kontrollieren"], ["Drehzahlsensor", "SPS-Eingang", "Sensorleitung"], "Mittel", )) if len(letzte_drehzahlen) >= 2 and max(letzte_drehzahlen[-2:] + [drehzahl]) - min(letzte_drehzahlen[-2:] + [drehzahl]) > normalwerte["Drehzahl"] * 0.15: fehler.append(erstelle_elektronikfehler( "Drehzahlsensor", "Drehzahlsignal instabil", ["Wackelkontakt", "Sensorabstand zu gross", "EMV-Stoerung", "Frequenzumrichter-Ausgang instabil"], ["Sensorposition pruefen", "Kabelschirmung pruefen", "Klemmen nachziehen", "FU-Fehlerspeicher pruefen"], ["Drehzahlsensor", "Sensorleitung", "Frequenzumrichter"], "Mittel", )) if strom == 0 and drehzahl > 0: fehler.append(erstelle_elektronikfehler( "Stromsensor", "Stromsensor liefert kein plausibles Signal", ["Stromsensor defekt", "Messwandler falsch angeschlossen", "Analogeingang defekt", "falsche Skalierung"], ["Stromsensor pruefen", "Messwandler vergleichen", "Analogeingang messen", "Skalierung im Programm kontrollieren"], ["Stromsensor", "Analogeingang / SPS-Modul", "Sensorleitung"], "Hoch", )) if strom > normalwerte["Stromaufnahme"] * SETTINGS["strom_extrem_faktor"]: fehler.append(erstelle_elektronikfehler( "Stromsensor", "Stromaufnahme extrem hoch", ["echter Ueberlastfall", "Messwertfehler", "falsche Skalierung", "Kurzschluss / Wicklungsproblem"], ["Motorstrom mit Messzange gegenpruefen", "Umrichterwert vergleichen", "Skalierung pruefen", "Motorwicklung pruefen"], ["Stromsensor", "Frequenzumrichter", "Motor"], "Hoch", )) if len(letzte_stroeme) >= 2 and max(letzte_stroeme[-2:] + [strom]) - min(letzte_stroeme[-2:] + [strom]) > normalwerte["Stromaufnahme"] * 0.5: fehler.append(erstelle_elektronikfehler( "Stromsensor", "Stromsignal instabil", ["Wackelkontakt", "defekter Messwandler", "EMV-Stoerung", "lose Klemme"], ["Klemmen pruefen", "Messwandler pruefen", "Schirmung und Erdung pruefen", "Signalverlauf beobachten"], ["Stromsensor", "Sensorleitung", "Analogeingang / SPS-Modul"], "Mittel", )) strom_hoch = berechne_abweichung_prozent(normalwerte["Stromaufnahme"], strom) >= SETTINGS["abweichung_gelb_prozent"] drehzahl_niedrig = berechne_abweichung_prozent(normalwerte["Drehzahl"], drehzahl) <= -SETTINGS["abweichung_gelb_prozent"] drehzahl_abweichung = abs(berechne_abweichung_prozent(normalwerte["Drehzahl"], drehzahl)) >= SETTINGS["abweichung_gelb_prozent"] if drehzahl_niedrig and strom_hoch: fehler.append(erstelle_elektronikfehler( "Frequenzumrichter", "Antrieb laeuft unter Last / moeglicher Ueberlastfall", ["mechanische Last zu hoch", "FU begrenzt Drehmoment", "Rampe / Parameter falsch", "Motor wird gebremst"], ["FU-Fehlerspeicher pruefen", "Sollwert/Istwert vergleichen", "Parameter pruefen", "Mechanik auf Schwergaengigkeit pruefen"], ["Frequenzumrichter", "Motor", "Mechanik"], "Hoch", )) elif drehzahl_abweichung and not strom_hoch: fehler.append(erstelle_elektronikfehler( "Frequenzumrichter", "Drehzahlabweichung ohne Lastanstieg", ["Sollwertfehler", "Regelungsfehler", "Drehzahlsensorfehler", "FU-Parameter falsch"], ["Sollwert pruefen", "FU-Ausgangsfrequenz pruefen", "SPS-Signal pruefen", "Sensor pruefen"], ["Frequenzumrichter", "Drehzahlsensor", "SPS-Signal"], "Mittel", )) if len(fehler) >= 3: fehler.append(erstelle_elektronikfehler( "SPS-/Signalfehler", "Mehrere Sensorwerte gleichzeitig unplausibel", ["24-V-Versorgung instabil", "Analogkarte fehlerhaft", "Masseproblem", "Kommunikationsproblem"], ["24-V-Versorgung messen", "SPS-Eingangskarte pruefen", "Klemmen und Masseverbindung pruefen", "Diagnosepuffer / Fehlermeldungen pruefen"], ["24-V-Versorgung", "SPS-Modul", "Klemmen", "Masseverbindung"], "Hoch", )) return fehler def berechne_wartungsstatus(motor): """Bewertet Betriebsstunden gegen das Wartungsintervall.""" reststunden = motor["wartungsintervall_stunden"] - motor["betriebsstunden"] if reststunden <= 0: status = "Wartung ueberfaellig" elif reststunden < 100: status = "Wartung bald faellig" elif reststunden <= 500: status = "Wartung einplanen" else: status = "Unkritisch" return { "betriebsstunden": motor["betriebsstunden"], "wartungsintervall_stunden": motor["wartungsintervall_stunden"], "reststunden_bis_wartung": reststunden, "wartungsstatus": status, } def berechne_energie_mehrkosten(motor, ergebnisse): """Schaetzt Energie-Mehrkosten auf Basis positiver Stromabweichung.""" strom_ergebnis = next( ergebnis for ergebnis in ergebnisse if ergebnis["messwert"] == "Stromaufnahme" ) stromabweichung = max(strom_ergebnis["abweichung_prozent"], 0) mehrleistung_kw = motor["nennleistung_kw"] * stromabweichung / 100 mehrkosten_tag = mehrleistung_kw * 24 * STROMPREIS_EURO_KWH mehrkosten_monat = mehrkosten_tag * 30 return { "hinweis": "Demo-Schaetzung auf Basis der Stromabweichung", "stromabweichung_prozent": round(stromabweichung, 1), "mehrleistung_kw": round(mehrleistung_kw, 2), "mehrkosten_pro_tag": round(mehrkosten_tag, 2), "mehrkosten_pro_monat": round(mehrkosten_monat, 2), "strompreis_euro_kwh": STROMPREIS_EURO_KWH, } def berechne_fehlerwahrscheinlichkeiten(ergebnisse): """Erstellt eine einfache Demo-Diagnose nach Baugruppen.""" wahrscheinlichkeiten = {} def addiere(name, wert): wahrscheinlichkeiten[name] = min(wahrscheinlichkeiten.get(name, 0) + wert, 100) werte = {ergebnis["messwert"]: ergebnis for ergebnis in ergebnisse} if werte["Vibration"]["status"] != "Gruen": addiere("Lagerproblem", 40) addiere("Unwucht", 30) addiere("Befestigung", 20) if werte["Stromaufnahme"]["abweichung_prozent"] >= SETTINGS["abweichung_gelb_prozent"]: addiere("Ueberlast", 30) addiere("Mechanischer Widerstand", 35) addiere("Frequenzumrichter", 15) if werte["Temperatur"]["status"] != "Gruen": addiere("Kuehlung", 30) addiere("Ueberlast", 25) addiere("Lagererwaermung", 20) if werte["Drehzahl"]["abweichung_prozent"] <= -SETTINGS["abweichung_gelb_prozent"]: addiere("Ueberlast", 25) addiere("Frequenzumrichter", 30) addiere("Regelung/Sollwert", 20) if not wahrscheinlichkeiten: return [{"baugruppe": "Keine auffaellige Baugruppe", "wahrscheinlichkeit": 0}] return [ {"baugruppe": name, "wahrscheinlichkeit": wert} for name, wert in sorted(wahrscheinlichkeiten.items(), key=lambda item: item[1], reverse=True) ] def markiere_betroffene_bauteile(bauteile, ergebnisse, elektronische_fehler): """Markiert Bauteile, die zur aktuellen Auffaelligkeit passen.""" markierungen = [] auffaellig = { ergebnis["messwert"]: ergebnis["status"] != "Gruen" for ergebnis in ergebnisse } elektronische_kategorien = [fehler["kategorie"] for fehler in elektronische_fehler] for bauteil in bauteile: text = f"{bauteil['gruppe']} {bauteil['bezeichnung']} {bauteil['typ']}".lower() betroffen = False if auffaellig.get("Temperatur") and "temperatur" in text: betroffen = True if auffaellig.get("Vibration") and any(wort in text for wort in ["vibration", "schwingung", "lager", "kupplung"]): betroffen = True if auffaellig.get("Drehzahl") and any(wort in text for wort in ["drehzahl", "frequenzumrichter", "fu"]): betroffen = True if auffaellig.get("Stromaufnahme") and any(wort in text for wort in ["strom", "frequenzumrichter", "motor"]): betroffen = True if "Temperatursensor" in elektronische_kategorien and "temperatur" in text: betroffen = True if "Vibrationssensor" in elektronische_kategorien and any(wort in text for wort in ["vibration", "schwingung"]): betroffen = True if "Drehzahlsensor" in elektronische_kategorien and "drehzahl" in text: betroffen = True if "Stromsensor" in elektronische_kategorien and "strom" in text: betroffen = True if "Frequenzumrichter" in elektronische_kategorien and any(wort in text for wort in ["frequenzumrichter", "fu"]): betroffen = True bauteil_mit_status = dict(bauteil) bauteil_mit_status["betroffen"] = betroffen markierungen.append(bauteil_mit_status) return markierungen def ermittle_trendrichtung(letzte_messungen): """Bewertet, ob die letzten Messungen steigen, stabil bleiben oder fallen.""" if len(letzte_messungen) < 3: return "noch zu wenig Daten" erste_haelfte = letzte_messungen[:len(letzte_messungen) // 2] zweite_haelfte = letzte_messungen[len(letzte_messungen) // 2:] durchschnitt_alt = sum(erste_haelfte) / len(erste_haelfte) durchschnitt_neu = sum(zweite_haelfte) / len(zweite_haelfte) differenz = durchschnitt_neu - durchschnitt_alt if differenz > 2: return "steigend" if differenz < -2: return "fallend" return "stabil" def ermittle_simulierten_motorzustand(motor, simulation_state): """Ermittelt den wirksamen Motorzustand aus Standardliste oder Tastermodus.""" modus = simulation_state.get("anlagenmodus", "standard") betriebsart = simulation_state.get("betriebsart", "automatik") if betriebsart == "stoerung": return "kritisch" if motor["name"] == "Motor Foerderband 1" else "normal" if modus == "normal": return "normal" if modus == "auffaellig": return "warnbetrieb" if modus == "kritisch": if motor["name"] in ["Motor Foerderband 1", "Motor Luefter 3"]: return "kritisch" return "warnbetrieb" if modus == "pumpe_warnung": return "warnbetrieb" if motor["name"] == "Motor Pumpe 2" else "normal" if modus == "luefter_kritisch": return "kritisch" if motor["name"] == "Motor Luefter 3" else "normal" if modus == "sensorfehler": return "normal" if modus == "versorgungsfehler": return "warnbetrieb" if motor["name"] == "Motor Foerderband 1" else "normal" if modus == "400v_fehler": return "warnbetrieb" if motor["name"] == "Motor Foerderband 1" else "normal" if modus in ["fu_stoerung", "materialstau"]: return "warnbetrieb" if motor["name"] == "Motor Foerderband 1" else "normal" if modus == "wiederholfehler": return "warnbetrieb" if motor["name"] == "Motor Foerderband 1" else "normal" if modus in ["reparatur_demo", "pruefprotokoll_erforderlich", "meisterfreigabe"]: return "normal" return motor["zustand"] def simuliere_messwerte(motor, simulation_state): """Simuliert aktuelle Messwerte passend zum Motorzustand.""" aktuelle_werte = {} modus = simulation_state.get("anlagenmodus", "standard") zustand = ermittle_simulierten_motorzustand(motor, simulation_state) bereiche = SIMULATIONS_BEREICHE_WARNBETRIEB if zustand == "warnbetrieb" else SIMULATIONS_BEREICHE[zustand] for messwert_name, normalwert in motor["normalwerte"].items(): minimum, maximum = bereiche[messwert_name] faktor = random.uniform(minimum, maximum) aktuelle_werte[messwert_name] = round(normalwert * faktor, 2) if modus == "fu_stoerung" and motor["name"] == "Motor Foerderband 1": aktuelle_werte["Drehzahl"] = 0.0 aktuelle_werte["Stromaufnahme"] = round(motor["normalwerte"]["Stromaufnahme"] * random.uniform(0.95, 1.05), 2) aktuelle_werte["Zeitstempel"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") return aktuelle_werte def lese_historie(): """Liest vorhandene Langzeitdaten aus der CSV-Datei.""" historie = {} if not os.path.exists(HISTORIE_PFAD): return historie with open(HISTORIE_PFAD, mode="r", encoding="utf-8", newline="") as datei: reader = csv.DictReader(datei) for zeile in reader: motorname = zeile["Motorname"] durchschnitt = float(zeile["DurchschnittAbweichung"]) historie.setdefault(motorname, []).append(durchschnitt) historie.setdefault(f"{motorname}:Stromaufnahme", []).append(float(zeile["Stromaufnahme"])) historie.setdefault(f"{motorname}:Temperatur", []).append(float(zeile["Temperatur"])) historie.setdefault(f"{motorname}:Vibration", []).append(float(zeile["Vibration"])) historie.setdefault(f"{motorname}:Drehzahl", []).append(float(zeile["Drehzahl"])) return historie def lese_quittierungen(): """Liest gespeicherte Alarm-Quittierungen.""" if not os.path.exists(QUITTIERUNGEN_PFAD): return {} with open(QUITTIERUNGEN_PFAD, mode="r", encoding="utf-8") as datei: return json.load(datei) def speichere_quittierungen(quittierungen): """Speichert Alarm-Quittierungen.""" schreibe_json_datei(QUITTIERUNGEN_PFAD, quittierungen) def ist_alarm_quittiert(bericht, quittierungen): """Prueft, ob der aktuelle Alarm eines Motors quittiert wurde.""" quittierung = quittierungen.get(f"motor:{bericht['motorname']}") or quittierungen.get(bericht["motorname"]) return ( bericht["gesamtstatus"] != "Gruen" and quittierung is not None and quittierung.get("status") == bericht["gesamtstatus"] ) def aktualisiere_quittierungen(berichte): """Setzt Quittierungen zurueck, wenn ein Motor wieder gruen ist.""" quittierungen = lese_quittierungen() for bericht in berichte: for schluessel in [bericht["motorname"], f"motor:{bericht['motorname']}"]: if bericht["gesamtstatus"] == "Gruen" and schluessel in quittierungen: del quittierungen[schluessel] speichere_quittierungen(quittierungen) return quittierungen def reset_alarm_historie(): """Setzt die Alarm-Historie auf die Kopfzeile zurueck.""" with open(ALARM_HISTORIE_PFAD, mode="w", encoding="utf-8", newline="") as datei: writer = csv.writer(datei) writer.writerow(ALARM_HISTORIE_FELDNAMEN) def speichere_ereignis(motorname, bereich, status, meldungstyp, risikowert, meldung, quittiert): """Speichert ein Ereignis in alarm_history.csv.""" datei_existiert = os.path.exists(ALARM_HISTORIE_PFAD) with open(ALARM_HISTORIE_PFAD, mode="a", encoding="utf-8", newline="") as datei: writer = csv.DictWriter(datei, fieldnames=ALARM_HISTORIE_FELDNAMEN) if not datei_existiert: writer.writeheader() writer.writerow( { "Zeitstempel": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "Motorname": motorname, "Bereich": bereich, "Status": status, "Meldungstyp": meldungstyp, "Risikowert": risikowert, "Meldung": meldung, "Quittiert": "Ja" if quittiert else "Nein", } ) def lese_letzte_ereignisse(anzahl=10): """Liest die letzten Ereignisse fuer das Dashboard.""" if not os.path.exists(ALARM_HISTORIE_PFAD): return [] with open(ALARM_HISTORIE_PFAD, mode="r", encoding="utf-8", newline="") as datei: eintraege = list(csv.DictReader(datei)) return eintraege[-anzahl:] def reset_reparatur_historie(): """Setzt die Reparaturhistorie auf die Kopfzeile zurueck.""" with open(REPARATUR_HISTORIE_PFAD, mode="w", encoding="utf-8", newline="") as datei: writer = csv.writer(datei) writer.writerow(REPARATUR_HISTORIE_FELDNAMEN) def stelle_reparatur_historie_schema_sicher(): """Erweitert eine alte repair_history.csv kompatibel um neue Felder.""" if not os.path.exists(REPARATUR_HISTORIE_PFAD): reset_reparatur_historie() return with open(REPARATUR_HISTORIE_PFAD, mode="r", encoding="utf-8", newline="") as datei: reader = csv.DictReader(datei) if reader.fieldnames == REPARATUR_HISTORIE_FELDNAMEN: return eintraege = list(reader) with open(REPARATUR_HISTORIE_PFAD, mode="w", encoding="utf-8", newline="") as datei: writer = csv.DictWriter(datei, fieldnames=REPARATUR_HISTORIE_FELDNAMEN) writer.writeheader() for eintrag in eintraege: writer.writerow({ feldname: eintrag.get(feldname, "") for feldname in REPARATUR_HISTORIE_FELDNAMEN }) def speichere_reparatur_eintrag( motor, fehlerbild, massnahme, ergebnis, kommentar, mitarbeiter, schicht="", fehlerart="", mechanisches_fehlerbild="", elektrisches_fehlerbild="", mechanische_massnahme="", elektrische_massnahme="", teile_gewechselt="", ): """Speichert eine dokumentierte Reparatur- oder Pruefmassnahme.""" stelle_reparatur_historie_schema_sicher() datei_existiert = os.path.exists(REPARATUR_HISTORIE_PFAD) zeitstempel = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(REPARATUR_HISTORIE_PFAD, mode="a", encoding="utf-8", newline="") as datei: writer = csv.DictWriter(datei, fieldnames=REPARATUR_HISTORIE_FELDNAMEN) if not datei_existiert: writer.writeheader() writer.writerow( { "Zeitstempel": zeitstempel, "Motorname": motor["motorname"], "Bereich": motor["bereich"], "Fehlerbild": fehlerbild, "Durchgefuehrte_Massnahme": massnahme, "Ergebnis": ergebnis, "Kommentar": kommentar, "Mitarbeiter": mitarbeiter, "Aktueller_Status": motor["gesamtstatus"], "Risikowert": motor["risikowert"], "Schicht": schicht, "Fehlerart": fehlerart, "Mechanisches_Fehlerbild": mechanisches_fehlerbild, "Elektrisches_Fehlerbild": elektrisches_fehlerbild, "Mechanische_Massnahme": mechanische_massnahme, "Elektrische_Massnahme": elektrische_massnahme, "Teile_gewechselt": teile_gewechselt, } ) speichere_ereignis( motor["motorname"], motor["bereich"], motor["gesamtstatus"], motor["meldungstyp"], motor["risikowert"], f"Reparatur dokumentiert: {massnahme} - {ergebnis}", motor.get("alarm_quittiert", False), ) def lese_reparaturhistorie(): """Liest alle dokumentierten Reparaturen.""" if not os.path.exists(REPARATUR_HISTORIE_PFAD): return [] with open(REPARATUR_HISTORIE_PFAD, mode="r", encoding="utf-8", newline="") as datei: eintraege = list(csv.DictReader(datei)) for eintrag in eintraege: for feldname in REPARATUR_HISTORIE_FELDNAMEN: eintrag.setdefault(feldname, "") return eintraege def lese_letzte_reparaturen(anzahl=10): """Liest die letzten Reparatureintraege fuer das Dashboard.""" return lese_reparaturhistorie()[-anzahl:] def lese_reparaturen_fuer_motor(motorname): """Liest Reparatureintraege fuer einen bestimmten Motor.""" return [ eintrag for eintrag in lese_reparaturhistorie() if eintrag.get("Motorname") == motorname ] def ermittle_aktuelle_fehlerbilder(ergebnisse, elektronische_fehler): """Leitet einfache Fehlerbild-Bezeichnungen aus aktuellen Auffaelligkeiten ab.""" fehlerbilder = [] for ergebnis in ergebnisse: if ergebnis["status"] == "Gruen": continue if ergebnis["messwert"] == "Vibration": fehlerbilder.append("Vibration erhoeht") elif ergebnis["messwert"] == "Temperatur": fehlerbilder.append("Temperatur erhoeht") elif ergebnis["messwert"] == "Stromaufnahme": fehlerbilder.append("Stromaufnahme erhoeht") elif ergebnis["messwert"] == "Drehzahl": fehlerbilder.append("Drehzahl abweichend") for fehler in elektronische_fehler: if fehler["kategorie"] == "Temperatursensor": fehlerbilder.append("Temperatursensor unplausibel") elif fehler["kategorie"] == "Vibrationssensor": fehlerbilder.append("Vibrationssensor ohne Signal") elif fehler["kategorie"] == "Drehzahlsensor": fehlerbilder.append("Drehzahlsensor ohne Signal") elif fehler["kategorie"] == "Stromsensor": fehlerbilder.append("Stromsensor unplausibel") elif fehler["kategorie"] == "Frequenzumrichter": fehlerbilder.append("Frequenzumrichterfehler") elif fehler["kategorie"] == "SPS-/Signalfehler": fehlerbilder.append("SPS-/Signalfehler vermutet") return list(dict.fromkeys(fehlerbilder)) def analysiere_reparaturhistorie(motorname, aktuelle_fehlerbilder): """Erzeugt Hinweise aus wiederkehrenden Fehlern und bisherigen Massnahmen.""" eintraege = lese_reparaturen_fuer_motor(motorname) hinweise = [] letzte_massnahme = None bereits_gewechselt = [] gepruefte_bereiche = [] if eintraege: letzter = eintraege[-1] letzte_massnahme = ( f"{letzter.get('Zeitstempel', '')} - {letzter.get('Durchgefuehrte_Massnahme', '')} - " f"{letzter.get('Ergebnis', '')} - {letzter.get('Mitarbeiter', '')}" ) for fehlerbild in aktuelle_fehlerbilder: gleiche_fehler = [ eintrag for eintrag in eintraege if eintrag.get("Fehlerbild") == fehlerbild ] if len(gleiche_fehler) >= 2: hinweise.append( f"Wiederholungsfehler erkannt: {fehlerbild} trat bei {motorname} bereits {len(gleiche_fehler)}x auf." ) massnahmen = [eintrag.get("Durchgefuehrte_Massnahme", "") for eintrag in eintraege] if any("Lager gewechselt" in massnahme for massnahme in massnahmen): bereits_gewechselt.append("Lager") if "Vibration erhoeht" in aktuelle_fehlerbilder: hinweise.append( "Lager wurde bereits gewechselt - vor erneutem Tausch Kupplung, Ausrichtung, Befestigung und Unwucht pruefen." ) if any("Luefter/Kuehlung geprueft" in massnahme or "Lüfter/Kühlung geprüft" in massnahme for massnahme in massnahmen): gepruefte_bereiche.append("Luefter/Kuehlung") if "Temperatur erhoeht" in aktuelle_fehlerbilder: hinweise.append("Kuehlung wurde bereits geprueft - Ueberlast oder Lagererwaermung pruefen.") if any("Mechanik auf Schwergaengigkeit geprueft" in massnahme or "Mechanik auf Schwergängigkeit geprüft" in massnahme for massnahme in massnahmen): gepruefte_bereiche.append("Mechanik") if "Stromaufnahme erhoeht" in aktuelle_fehlerbilder: hinweise.append("Mechanik wurde bereits geprueft - Frequenzumrichter oder Motorzustand pruefen.") if aktuelle_fehlerbilder.count("Temperatursensor unplausibel") or any(e.get("Fehlerbild") == "Temperatursensor unplausibel" for e in eintraege): gleiche = [e for e in eintraege if e.get("Fehlerbild") == "Temperatursensor unplausibel"] if len(gleiche) >= 2: hinweise.append("Wiederholter Temperatursensorfehler - Leitung, Klemmen und Analogeingang pruefen, bevor Sensor erneut getauscht wird.") if any(e.get("Fehlerbild") == "Wackelkontakt vermutet" for e in eintraege): hinweise.append("Wiederkehrender Wackelkontakt - Klemmen, Steckverbinder und Kabelbewegung pruefen.") if any("Sensor gewechselt" in massnahme for massnahme in massnahmen): bereits_gewechselt.append("Sensor") hinweise.append("Sensor wurde bereits gewechselt - Fehlerquelle kann Leitung, Eingangskarte oder Skalierung sein.") return { "letzte_massnahme": letzte_massnahme or "Noch keine Reparatur dokumentiert", "hinweise": hinweise or ["Keine wiederkehrenden Reparaturmuster erkannt"], "bereits_gewechselte_teile": list(dict.fromkeys(bereits_gewechselt)) or ["Keine dokumentierten Teilewechsel"], "gepruefte_bereiche": list(dict.fromkeys(gepruefte_bereiche)) or ["Keine dokumentierten Pruefbereiche"], } def ermittle_nicht_nochmal_tauschen_warnungen(motorname, aktuelle_fehlerbilder): """Warnt, wenn ein Teil bei gleichem Fehler bereits gewechselt wurde.""" eintraege = lese_reparaturen_fuer_motor(motorname) massnahmen = " | ".join(eintrag.get("Durchgefuehrte_Massnahme", "") for eintrag in eintraege) teile = " | ".join(eintrag.get("Teile_gewechselt", "") for eintrag in eintraege) text = f"{massnahmen} | {teile}".lower() warnungen = [] if "Vibration erhoeht" in aktuelle_fehlerbilder and "lager" in text: warnungen.append({ "bauteil": "Lager", "letzte_massnahme": "Lager wurde bereits geprueft oder gewechselt", "hinweis": "Achtung: Das Lager wurde bereits gewechselt/geprueft. Vor erneutem Austausch zuerst alternative Ursachen pruefen.", "alternative_pruefungen": ["Kupplung", "Ausrichtung", "Befestigung", "Unwucht", "Sensorbefestigung"], }) if "Temperatursensor unplausibel" in aktuelle_fehlerbilder and "sensor" in text: warnungen.append({ "bauteil": "Temperatursensor", "letzte_massnahme": "Sensor wurde bereits gewechselt oder geprueft", "hinweis": "Temperatursensor wurde bereits gewechselt/geprueft. Leitung, Klemmen, Analogkarte und Skalierung pruefen.", "alternative_pruefungen": ["Sensorleitung", "Klemmen", "Analogeingang", "Skalierung"], }) if "Drehzahl abweichend" in aktuelle_fehlerbilder and "frequenzumrichter" in text: warnungen.append({ "bauteil": "Frequenzumrichter", "letzte_massnahme": "FU wurde bereits geprueft", "hinweis": "FU wurde bereits geprueft. Sollwert, SPS-Signal, Sensorik und mechanische Last pruefen.", "alternative_pruefungen": ["Sollwert", "SPS-Signal", "Drehzahlsensor", "mechanische Last"], }) return warnungen def demo_schritt_aus_zeitpunkt(simulation_state): """Ermittelt den aktuellen Demo-Schritt fuer laufenden Automatikbetrieb.""" betriebsart = simulation_state.get("betriebsart", "automatik") modus = simulation_state.get("anlagenmodus", "normal") if betriebsart in ["handbetrieb", "wartung", "gestoppt"]: return 1 if betriebsart == "handbetrieb" else 0 if betriebsart == "stoerung" or modus in ["kritisch"]: return 99 if modus == "materialstau": return 6 if modus == "fu_stoerung": return 5 if modus == "400v_fehler": return 2 if modus == "sensorfehler": return 4 schrittzeit_s = 4 basis_schritte = list(range(0, 12)) schritt = basis_schritte[int(datetime.now().timestamp() // schrittzeit_s) % len(basis_schritte)] return 4 if schritt == 0 and modus in ["normal", "meisterfreigabe"] else schritt def ermittle_materialfluss_detail(schrittkette, simulation_state): """ Erstellt eine Demo-Logik fuer den Materialfluss. Keine echte Anlagensteuerung, nur Visualisierung fuer Dashboard/HMI. """ aktueller_schritt = schrittkette.get("aktueller_schritt", 0) stoerung = schrittkette.get("stoerung", False) betriebsart = simulation_state.get("betriebsart", "automatik") anlagenmodus = simulation_state.get("anlagenmodus", "normal") material = { "status": "bereit", "position": "grundstellung", "teile_auf_band": 0, "teil_bei_s1": False, "teil_bei_s2": False, "teil_in_pruefung": False, "stau": False, "richtung": "links_nach_rechts", "beschreibung": "Anlage bereit, kein aktiver Materialtransport.", "naechster_erwarteter_sensor": "-", "letzte_position": "-", "ziel_position": "-", } if betriebsart in ["handbetrieb", "wartung", "gestoppt"]: material.update({ "status": "steht", "position": "pausiert", "teile_auf_band": 1, "beschreibung": "Materialfluss steht wegen Betriebsart.", }) return material if anlagenmodus == "400v_fehler": material.update({ "status": "steht", "position": "vor_start", "teile_auf_band": 0, "beschreibung": "Materialfluss steht, weil der 400-V-Leistungskreis den Motorstart sperrt.", "naechster_erwarteter_sensor": "S1", "letzte_position": "-", "ziel_position": "Startfreigabe", }) return material if anlagenmodus == "fu_stoerung": material.update({ "status": "steht", "position": "m1_start", "teile_auf_band": 1, "beschreibung": "M1 startet nicht: Startsignal liegt an, aber FU-Stoerung/Drehzahlrueckmeldung fehlt.", "naechster_erwarteter_sensor": "S2", "letzte_position": "S1", "ziel_position": "S2", }) return material if stoerung or anlagenmodus in ["kritisch", "materialstau"]: material.update({ "status": "steht", "position": "stau_zwischen_s1_und_s2", "teile_auf_band": 3, "teil_bei_s1": True, "teil_bei_s2": False, "stau": True, "beschreibung": "Materialfluss gestoert. Teil erreicht S2 nicht rechtzeitig.", "naechster_erwarteter_sensor": "S2", "letzte_position": "S1", "ziel_position": "S2", }) return material if aktueller_schritt in [0, 1, 2, 3]: material.update({ "status": "bereit", "position": "vor_start", "beschreibung": "Startfreigabe, Versorgung und SPS/FU werden geprueft.", "naechster_erwarteter_sensor": "S1", }) elif aktueller_schritt == 4: material.update({ "status": "aktiv", "position": "s1", "teile_auf_band": 1, "teil_bei_s1": True, "beschreibung": "S1 erkennt ein Teil am Materialeingang.", "naechster_erwarteter_sensor": "S2", "letzte_position": "S1", "ziel_position": "Foerderband", }) elif aktueller_schritt in [5, 6]: material.update({ "status": "aktiv", "position": "zwischen_s1_und_s2", "teile_auf_band": 3, "beschreibung": "M1 foerdert Teile ueber das Band Richtung S2.", "naechster_erwarteter_sensor": "S2", "letzte_position": "S1", "ziel_position": "S2", }) elif aktueller_schritt == 7: material.update({ "status": "aktiv", "position": "s2", "teile_auf_band": 2, "teil_bei_s2": True, "beschreibung": "S2 erkennt das Teil am Materialausgang.", "naechster_erwarteter_sensor": "Pruefung", "letzte_position": "S2", "ziel_position": "Pruefung", }) elif aktueller_schritt in [8, 9]: material.update({ "status": "aktiv", "position": "pruefung", "teile_auf_band": 1, "teil_in_pruefung": True, "beschreibung": "Teil befindet sich in der Pruefstation.", "naechster_erwarteter_sensor": "Ausgang", "letzte_position": "Pruefung", "ziel_position": "Ausgang", }) elif aktueller_schritt in [10, 11]: material.update({ "status": "aktiv", "position": "ausgang", "teile_auf_band": 1, "beschreibung": "Teil wird ausgeschleust und uebergeben.", "naechster_erwarteter_sensor": "S1", "letzte_position": "Ausgang", "ziel_position": "Naechstes Teil", }) return material def diagnostiziere_motorstart(sps_io, schrittkette): """Demo-Diagnose: Warum startet Motor M1 nicht?""" diagnose = { "aktiv": False, "titel": "Motorstart unauffaellig", "ursache": "-", "pruefschritte": [], "empfehlung": "Keine Motorstart-Stoerung erkannt.", } digitale_eingaenge = {x["adresse"]: x for x in sps_io.get("eingänge_digital", [])} digitale_ausgaenge = {x["adresse"]: x for x in sps_io.get("ausgänge_digital", [])} analoge_eingaenge = {x["adresse"]: x for x in sps_io.get("eingänge_analog", [])} q_start = digitale_ausgaenge.get("Q0.1", {}).get("wert", 0) q_freigabe = digitale_ausgaenge.get("Q0.0", {}).get("wert", 0) motorschutz_ok = digitale_eingaenge.get("I0.1", {}).get("wert", 1) fu_stoerung = digitale_eingaenge.get("I0.7", {}).get("wert", 0) drehzahl = float(analoge_eingaenge.get("IW70", {}).get("wert", 0) or 0) if q_start == 1 and q_freigabe == 1 and drehzahl <= 50: diagnose["aktiv"] = True diagnose["titel"] = "Motor M1 startet nicht trotz Startsignal" if motorschutz_ok == 0: diagnose["ursache"] = "Motorschutz nicht OK" diagnose["pruefschritte"] = ["Motorschutz I0.1 pruefen", "Motorstrom messen", "Mechanik auf Blockade pruefen"] diagnose["empfehlung"] = "Motorschutz nicht einfach zuruecksetzen, zuerst Ursache pruefen." elif fu_stoerung == 1: diagnose["ursache"] = "FU-Stoerung aktiv" diagnose["pruefschritte"] = ["I0.7 FU-Stoerung pruefen", "FU-Fehlerspeicher auslesen", "Q0.1 Startsignal am FU pruefen"] diagnose["empfehlung"] = "FU-Fehlerspeicher auslesen und Ursache beseitigen." else: diagnose["ursache"] = "Startsignal aktiv, aber keine Drehzahl" diagnose["pruefschritte"] = ["Drehzahlsensor I0.5 / IW70 pruefen", "Motor mechanisch pruefen", "FU-Ausgang und Motorleitung pruefen"] diagnose["empfehlung"] = "Drehzahlrueckmeldung, Mechanik und FU-Ausgang pruefen." return diagnose def begrenze_prozent(wert, minimum=0, maximum=120): """Begrenzt Prozentwerte fuer stabile Demo-Anzeigen.""" return round(max(minimum, min(maximum, wert)), 1) def berechne_oee_light(produktion, schrittkette): """Berechnet eine einfache Demo-OEE aus vorhandenen Produktionsdaten.""" soll = produktion.get("sollleistung_pro_stunde", 0) or 0 ist = produktion.get("aktuelle_leistung_pro_stunde", 0) or 0 produziert = produktion.get("produzierte_teile_heute", 0) or 0 gutteile = produktion.get("gutteile", 0) or 0 leistung = begrenze_prozent((ist / soll) * 100) if soll > 0 else 0 qualitaet = begrenze_prozent((gutteile / produziert) * 100) if produziert > 0 else 100 verfuegbarkeit = 100 if produktion.get("produktionsverlust_aktiv") or schrittkette.get("stoerung"): verfuegbarkeit = max(0, 100 - (produktion.get("leistungsverlust_prozent", 0) or 0)) verfuegbarkeit = begrenze_prozent(verfuegbarkeit, 0, 100) oee = begrenze_prozent((verfuegbarkeit * leistung * qualitaet) / 10000) return { "verfuegbarkeit_prozent": verfuegbarkeit, "leistung_prozent": leistung, "qualitaet_prozent": qualitaet, "oee_prozent": oee, "hinweis": "Demo-OEE auf Basis simulierter Produktionsdaten.", } def ermittle_engpassanalyse(produktion, motoren, schrittkette, materialfluss_detail, energieversorgung, aktive_fehlercodes): """Leitet eine einfache Engpassanalyse aus Materialfluss, Fehlercodes und Motorstatus ab.""" fehlercodes = set(aktive_fehlercodes or []) if energieversorgung.get("versorgung_400v", {}).get("status") != "OK": return { "engpass_aktiv": True, "komponente": "400-V-Leistungskreis / Antriebsversorgung", "ursache": "Motorstart ist durch fehlenden oder gesperrten Leistungskreis blockiert.", "empfehlung": "Hauptschalter, Sicherungen, FU-Einspeisung und 400-V-Messwerte pruefen.", } if materialfluss_detail.get("stau") or "FB1-ERR-008" in fehlercodes: return { "engpass_aktiv": True, "komponente": "Materialfluss zwischen S1 und S2", "ursache": "Teil erreicht S2 nicht rechtzeitig.", "empfehlung": "Bandbereich zwischen S1 und S2 sowie Sensor S2 pruefen.", } if "FB1-ERR-004" in fehlercodes or "FB1-ERR-001" in fehlercodes: return { "engpass_aktiv": True, "komponente": "SPS/FU / M1 Foerderbandantrieb", "ursache": "FU-Stoerung oder Motorstartproblem.", "empfehlung": "FU-Fehlerspeicher, Q0.1 und Drehzahlrueckmeldung pruefen.", } if energieversorgung.get("versorgung_24v", {}).get("status") != "OK": return { "engpass_aktiv": True, "komponente": "24-V-Steuerkreis / Sensorik", "ursache": "Sensorwerte koennen durch 24-V-Problem unzuverlaessig sein.", "empfehlung": "24-V-Netzteil, 0V, Klemmen und Sensorversorgung pruefen.", } for motor in motoren: status = motor.get("gesamtstatus") if status in ["Rot", "Gelb"]: return { "engpass_aktiv": True, "komponente": motor.get("motorname", "-"), "ursache": motor.get("produktionsauswirkung", {}).get("auswirkung", "Motor beeinflusst Produktion."), "empfehlung": motor.get("produktionsauswirkung", {}).get("empfehlung", "Motor pruefen."), } return { "engpass_aktiv": False, "komponente": "-", "ursache": "Kein Engpass erkannt", "empfehlung": "Keine Massnahme erforderlich", } def ermittle_verlustanalyse(produktion, engpassanalyse, materialfluss_detail, aktive_fehlercodes): """Erklaert den Produktionsverlust mit Ursache, Komponente und Massnahme.""" fehlercodes = set(aktive_fehlercodes or []) if not produktion.get("produktionsverlust_aktiv") and not engpassanalyse.get("engpass_aktiv"): return { "aktiv": False, "ursache": "Kein Produktionsverlust aktiv", "betroffene_komponente": "-", "fehlercode": "-", "empfohlene_massnahme": "Keine Massnahme erforderlich", } if "400-V" in engpassanalyse.get("komponente", ""): return { "aktiv": True, "ursache": "400-V-Leistungskreis / Motorstart gesperrt", "betroffene_komponente": "400 V / SPS/FU / M1", "fehlercode": "FB1-ERR-004" if "FB1-ERR-004" in fehlercodes else "-", "empfohlene_massnahme": "400-V-Versorgung, Sicherungen und FU-Einspeisung pruefen.", } if materialfluss_detail.get("stau") or "FB1-ERR-008" in fehlercodes: return { "aktiv": True, "ursache": "Materialstau zwischen S1 und S2", "betroffene_komponente": "Foerderband M1 / Sensor S2", "fehlercode": "FB1-ERR-008", "empfohlene_massnahme": "Bandbereich und Sensor S2 pruefen.", } if "FB1-ERR-004" in fehlercodes or "FB1-ERR-001" in fehlercodes: return { "aktiv": True, "ursache": "Motorstart / FU-Stoerung", "betroffene_komponente": "SPS/FU / M1", "fehlercode": "FB1-ERR-004" if "FB1-ERR-004" in fehlercodes else "FB1-ERR-001", "empfohlene_massnahme": "FU-Fehlerspeicher, Q0.1 und Drehzahlrueckmeldung pruefen.", } return { "aktiv": True, "ursache": engpassanalyse.get("ursache", "Produktionsleistung reduziert"), "betroffene_komponente": engpassanalyse.get("komponente", "-"), "fehlercode": next(iter(fehlercodes), "-") if fehlercodes else "-", "empfohlene_massnahme": engpassanalyse.get("empfehlung", "Ursache pruefen."), } def erweitere_produktionsdaten(produktion, motoren, schrittkette, materialfluss_detail, energieversorgung, aktive_fehlercodes): """Ergaenzt Produktion um OEE-Light, Engpass, Tagesziel und Verlustanalyse.""" engpass = ermittle_engpassanalyse(produktion, motoren, schrittkette, materialfluss_detail, energieversorgung, aktive_fehlercodes) oee = berechne_oee_light(produktion, schrittkette) ziel_teile = 1000 produziert = produktion.get("produzierte_teile_heute", 0) or 0 fortschritt = begrenze_prozent((produziert / ziel_teile) * 100, 0, 100) verlustanalyse = ermittle_verlustanalyse(produktion, engpass, materialfluss_detail, aktive_fehlercodes) leistungsverlust = produktion.get("leistungsverlust_prozent", 0) or 0 if produktion.get("produktionsverlust_aktiv") or schrittkette.get("stoerung") or leistungsverlust >= 50: ampelfarbe = "Rot" ampeltext = "Produktion kritisch reduziert" elif engpass.get("engpass_aktiv") or leistungsverlust >= 10: ampelfarbe = "Gelb" ampeltext = "Produktion eingeschraenkt / beobachten" else: ampelfarbe = "Gruen" ampeltext = "Produktion laeuft im Sollbereich" produktion.update({ "oee_light": oee, "engpassanalyse": engpass, "tagesziel": { "ziel_teile": ziel_teile, "fortschritt_prozent": fortschritt, }, "produktionsampel": { "status": ampelfarbe, "text": ampeltext, }, "verlustanalyse": verlustanalyse, "schrittkette_kurz": { "aktueller_schritt": schrittkette.get("aktueller_schritt"), "aktueller_schritt_name": schrittkette.get("aktueller_schritt_name"), "materialfluss": materialfluss_detail.get("status", schrittkette.get("materialfluss")), "startfreigabe": schrittkette.get("startfreigabe_status", "OK"), }, }) return produktion def erstelle_signalwege(datenqualitaet, energieversorgung, sensoren, sps_io, aktive_fehlercodes, diagnose_motorstart=None): """Erstellt Industrie-4.0-Diagnoseketten fuer Sensorik, Messwerte und Motorstart.""" signalwege = [] dq_status = datenqualitaet.get("status", "Gut") dq_score = datenqualitaet.get("diagnose_sicherheit_prozent", 90) v24_status = energieversorgung.get("versorgung_24v", {}).get("status", "OK") fehlercodes = set(aktive_fehlercodes or []) sensor_map = {sensor.get("id"): sensor for sensor in sensoren} digitale = {x["adresse"]: x for x in sps_io.get("eingänge_digital", []) + sps_io.get("ausgänge_digital", [])} analoge = {x["adresse"]: x for x in sps_io.get("eingänge_analog", []) + sps_io.get("ausgänge_analog", [])} diagnose_motorstart = diagnose_motorstart or {} def status_fuer(relevante_codes=None, adresse=None, sensor_id=None): relevante_codes = relevante_codes or [] if any(code in fehlercodes for code in relevante_codes): return "Fehler" if any(code in ["FB1-ERR-001", "FB1-ERR-004", "FB1-ERR-008"] for code in relevante_codes if code in fehlercodes) else "Warnung" if sensor_id and sensor_map.get(sensor_id, {}).get("status") != "OK": return sensor_map.get(sensor_id, {}).get("status", "Warnung") if sensor_id: if v24_status != "OK": return "Warnung" return "OK" io = digitale.get(adresse) or analoge.get(adresse) or {} if io.get("status") in ["Fehler", "Warnung"]: return io.get("status") return "OK" def kette(names, status="OK", warn_bis=0): return [ { "name": name, "status": status if index < warn_bis else "OK", "hinweis": "Pruefen" if index < warn_bis and status != "OK" else "plausibel", } for index, name in enumerate(names) ] def add(name, messwert, adresse, codes, pruefstelle, pruefung, chain_names, sensor_id=None): status = status_fuer(codes, adresse, sensor_id) warn_bis = 3 if status != "OK" else 0 if v24_status != "OK" and sensor_id: pruefstelle = "24-V-Versorgung / 0V / Klemmen zuerst pruefen" signalwege.append({ "name": name, "messwert": messwert, "sps_adresse": adresse, "status": status, "datenqualitaet": dq_status, "diagnose_sicherheit_prozent": dq_score, "vermutete_pruefstelle": "keine Auffaelligkeit" if status == "OK" else pruefstelle, "fehlercodes": codes, "kette": kette(chain_names, status, warn_bis), "empfohlene_pruefung": pruefung, }) add("Temperatursensor", "Temperatur", "IW66", ["FB1-ERR-003"], "PT100 Sensorleitung / Klemme X12 / IW66", ["Sensorleitung pruefen", "Klemme X12 pruefen", "PT100-Widerstand messen", "Analogeingang IW66 pruefen", "Skalierung pruefen"], ["PT100 Sensor", "Sensorleitung", "Klemme X12", "Analogeingang IW66", "SPS-Auswertung", "Dashboard"]) add("Vibrationssensor", "Vibration", "IW68", ["FB1-ERR-002"], "Sensorbefestigung / Schirmung / Klemme X13", ["Sensorbefestigung pruefen", "Kabel / Schirmung pruefen", "Klemme X13 pruefen", "4-20-mA-Signal pruefen", "Vergleichsmessung durchfuehren"], ["Vibrationssensor 4-20 mA", "Sensorleitung", "Klemme X13", "Analogeingang IW68", "SPS-Auswertung", "Dashboard"]) add("Drehzahlsensor", "Drehzahl", "I0.5 / IW70", ["FB1-ERR-001"], "Drehzahlsensor / Klemme X15 / I0.5 / IW70", ["Sensorabstand pruefen", "Impulsgeber pruefen", "I0.5 pruefen", "IW70 Skalierung pruefen", "Motor/FU Zustand vergleichen"], ["Drehzahlsensor", "Klemme X15", "Digitaleingang I0.5", "SPS-Zaehler", "Skalierung IW70", "Dashboard"]) add("Stromsensor", "Stromaufnahme", "IW64", ["FB1-ERR-006"], "Messwandler / Klemme X11 / IW64", ["Messwandler pruefen", "Klemme X11 pruefen", "IW64 pruefen", "Skalierung pruefen", "Strommessung mit Zange vergleichen"], ["Messwandler 4-20 mA", "Klemme X11", "Analogeingang IW64", "SPS-Auswertung", "Skalierung", "Dashboard"]) add("Sensor S1", "Materialeingang", "I0.2", ["FB1-ERR-005"], "S1 / Leitung / SPS-Eingang I0.2", ["Sensor S1 reinigen", "24 V messen", "Klemme pruefen", "I0.2 pruefen", "Materialzufuhr pruefen"], ["S1 Sensor", "Leitung", "Klemme X20", "Digitaleingang I0.2", "SPS-Auswertung", "Dashboard"], "FB1-S1") add("Sensor S2", "Materialausgang", "I0.3", ["FB1-ERR-005", "FB1-ERR-008"], "S2 / Bandbereich zwischen S1 und S2 / I0.3", ["Bandbereich zwischen S1 und S2 pruefen", "Sensor S2 reinigen", "Klemme pruefen", "I0.3 pruefen", "Materialstau beseitigen"], ["S2 Sensor", "Leitung", "Klemme X21", "Digitaleingang I0.3", "SPS-Auswertung", "Dashboard"], "FB1-S2") add("Sensor S3", "Bandlauf", "I0.4", ["FB1-ERR-007"], "Bandschieflaufsensor / Bandkante / I0.4", ["Bandlauf pruefen", "Sensorabstand pruefen", "Klemme pruefen", "I0.4 pruefen", "Bandfuehrung kontrollieren"], ["S3 Bandlauf", "Leitung", "Klemme X22", "Digitaleingang I0.4", "SPS-Auswertung", "Dashboard"], "FB1-S3") motorstart_status = "Fehler" if diagnose_motorstart.get("aktiv") else status_fuer(["FB1-ERR-001", "FB1-ERR-004"], "Q0.1") signalwege.append({ "name": "Motorstart", "messwert": "Startfreigabe / Drehzahl", "sps_adresse": "Q0.0 / Q0.1 / I0.7 / IW70", "status": motorstart_status, "datenqualitaet": dq_status, "diagnose_sicherheit_prozent": dq_score, "vermutete_pruefstelle": diagnose_motorstart.get("ursache") if diagnose_motorstart.get("aktiv") else "Startfreigabe / 400 V / FU pruefen" if motorstart_status != "OK" else "keine Auffaelligkeit", "fehlercodes": ["FB1-ERR-001", "FB1-ERR-004"], "kette": [ {"name": "Dashboard / Steuerlogik", "status": "OK", "hinweis": "Startanforderung"}, {"name": "Startfreigabe", "status": "OK" if motorstart_status == "OK" else "Warnung", "hinweis": "Freigabekette"}, {"name": "Q0.0 Motorfreigabe", "status": "OK", "hinweis": f"Wert {digitale.get('Q0.0', {}).get('wert', '-')}"}, {"name": "Q0.1 FU Start", "status": "OK", "hinweis": f"Wert {digitale.get('Q0.1', {}).get('wert', '-')}"}, {"name": "Frequenzumrichter", "status": "Fehler" if digitale.get("I0.7", {}).get("wert") else "OK", "hinweis": f"I0.7 {digitale.get('I0.7', {}).get('wert', '-')}"}, {"name": "Drehzahl I0.5/IW70", "status": "Fehler" if diagnose_motorstart.get("aktiv") else "OK", "hinweis": f"IW70 {analoge.get('IW70', {}).get('wert', '-')}"}, {"name": "Dashboard", "status": motorstart_status, "hinweis": diagnose_motorstart.get("empfehlung", "plausibel")}, ], "empfohlene_pruefung": diagnose_motorstart.get("pruefschritte") or ["Startfreigabe pruefen", "Q0.0 pruefen", "Q0.1 pruefen", "I0.7 FU-Stoerung pruefen", "Drehzahlrueckmeldung pruefen"], }) return signalwege def trenne_mechanische_elektrische_diagnose(ergebnisse, kombinationsdiagnosen, elektronische_fehler): """Ordnet Auffaelligkeiten grob Mechanik, Elektrik und Sensorik zu.""" mechanik = [] elektrik = [] sensorik = [] werte = {ergebnis["messwert"]: ergebnis for ergebnis in ergebnisse} if werte["Vibration"]["status"] != "Gruen": mechanik.append("Vibration auffaellig: Lager, Unwucht, Befestigung oder Kupplung pruefen") if werte["Stromaufnahme"]["abweichung_prozent"] >= SETTINGS["abweichung_gelb_prozent"]: mechanik.append("Stromaufnahme hoch: mechanische Last oder Schwergaengigkeit pruefen") elektrik.append("Stromaufnahme hoch: Umrichterwert und Strommessung gegenpruefen") if werte["Temperatur"]["status"] != "Gruen": mechanik.append("Temperatur auffaellig: Ueberlast, Kuehlung oder Lagererwaermung pruefen") if abs(werte["Drehzahl"]["abweichung_prozent"]) >= SETTINGS["abweichung_gelb_prozent"]: elektrik.append("Drehzahl abweichend: Frequenzumrichter, Sollwert und Regelung pruefen") for diagnose in kombinationsdiagnosen: mechanik.append(diagnose) for fehler in elektronische_fehler: sensorik.append(f"{fehler['kategorie']}: {fehler['diagnose']}") if sensorik and mechanik: gewerk = "Mechanik + Elektrik" elif sensorik or elektrik: gewerk = "Elektrik" elif mechanik: gewerk = "Mechanik" else: gewerk = "Beobachten" return { "mechanische_diagnose": mechanik or ["Keine eindeutige mechanische Auffaelligkeit erkannt."], "elektrische_diagnose": elektrik or ["Keine eindeutige elektrische Auffaelligkeit erkannt."], "sensor_signal_diagnose": sensorik or ["Keine elektrischen/sensorischen Auffaelligkeiten erkannt."], "empfohlenes_gewerk": gewerk, } def ist_sensor_quittiert(sensor, quittierungen): """Prueft, ob ein Sensoralarm quittiert wurde.""" schluessel = f"sensor:{sensor['id']}" quittierung = quittierungen.get(schluessel) return ( sensor["status"] != "OK" and quittierung is not None and quittierung.get("status") == sensor["status"] ) def simuliere_zusatzsensoren(berichte, quittierungen, simulation_state): """Simuliert einfache Foerderband-Sensoren fuer die HMI-Anlagenansicht.""" motor_fb1 = next((bericht for bericht in berichte if bericht["motorname"] == "Motor Foerderband 1"), None) sensoren = [] demo_schritt = demo_schritt_aus_zeitpunkt(simulation_state) modus = simulation_state.get("anlagenmodus", "standard") for basis in ZUSATZSENSOREN_FOERDERBAND: sensor = dict(basis) status = "OK" meldungstyp = "Keine Meldung" hinweis = "Signal plausibel" if basis["id"] == "FB1-S1": wert = "1" if demo_schritt == 4 else "0" hinweis = "S1 erkennt Teil am Eingang" if wert == "1" else "Kein Teil am Eingang" elif basis["id"] == "FB1-S2": wert = "1" if demo_schritt == 7 else "0" hinweis = "S2 erkennt Teil am Ausgang" if wert == "1" else "Kein Teil am Ausgang" else: wert = "1" hinweis = "Bandlauf OK" if modus in ["normal", "reparatur_demo", "pruefprotokoll_erforderlich", "meisterfreigabe"]: status = "OK" meldungstyp = "Keine Meldung" elif modus == "sensorfehler" and basis["id"] in ["FB1-S1", "FB1-S2"]: status = "Fehler" if basis["id"] == "FB1-S1" else "Warnung" meldungstyp = "Sensor-/Signalalarm" hinweis = "Demo-Sensorfehler: Signalweg bis SPS-Eingang pruefen" wert = "0" if basis["id"] == "FB1-S1" else "1" elif modus == "materialstau" and basis["id"] == "FB1-S1": status = "OK" meldungstyp = "Keine Meldung" hinweis = "S1 erkennt Material, aber S2 wird nicht erreicht" wert = "1" elif modus == "materialstau" and basis["id"] == "FB1-S2": status = "Fehler" meldungstyp = "Materialflussalarm" hinweis = "Materialstau moeglich: S1 erkennt Material, S2 meldet keinen Ausgang" wert = "0" elif modus == "versorgungsfehler" and basis["id"] != "FB1-S3": status = "Warnung" meldungstyp = "Sensorwarnung" hinweis = "24-V-Versorgung niedrig - Sensorversorgung und Klemmen pruefen" elif modus == "kritisch" and basis["id"] == "FB1-S3": status = "Fehler" meldungstyp = "Sensor-/Anlagenalarm" hinweis = "Bandlauf pruefen - Störung per Demo-Taster aktiv" wert = "0" elif basis["id"] == "FB1-S3" and motor_fb1 and motor_fb1["gesamtstatus"] == "Rot" and modus != "fu_stoerung": status = "Fehler" meldungstyp = "Sensor-/Anlagenalarm" hinweis = "Bandlauf pruefen - Motorstatus kritisch" wert = "0" elif basis["id"] == "FB1-S2" and motor_fb1 and motor_fb1["gesamtstatus"] == "Gelb": status = "Warnung" meldungstyp = "Sensorwarnung" hinweis = "Ausgangssignal beobachten" elif random.random() < 0.01: status = "Warnung" meldungstyp = "Sensorwarnung" hinweis = "Kurzzeitige Signalauffaelligkeit" sensor.update({ "status": status, "meldungstyp": meldungstyp, "letzter_wert": wert, "hinweis": hinweis, "alarm_quittiert": False, }) sensor["alarm_quittiert"] = ist_sensor_quittiert(sensor, quittierungen) sensoren.append(sensor) return sensoren def aktualisiere_sensor_quittierungen(sensoren, quittierungen): """Setzt Sensorquittierungen zurueck, wenn der Sensor wieder OK ist.""" for sensor in sensoren: schluessel = f"sensor:{sensor['id']}" if sensor["status"] == "OK" and schluessel in quittierungen: del quittierungen[schluessel] speichere_quittierungen(quittierungen) return quittierungen def finde_bericht(berichte, motorname): """Findet einen Motorbericht in der aktuellen Messrunde.""" return next((bericht for bericht in berichte if bericht["motorname"] == motorname), None) def finde_messwert(bericht, messwert_name): """Liest einen Messwert aus einem Motorbericht.""" for messwert in bericht.get("messwerte", []): if messwert["messwert"] == messwert_name: return messwert return {} def io_eintrag(adresse, bezeichnung, signaltyp, wert, status, hinweis, pruefung=None, einheit=""): """Erstellt einen einheitlichen SPS-I/O-Eintrag.""" return { "adresse": adresse, "bezeichnung": bezeichnung, "signaltyp": signaltyp, "wert": wert, "einheit": einheit, "status": status, "hinweis": hinweis, "empfohlene_pruefung": pruefung or ["Signal und Verdrahtung pruefen"], } def erstelle_energieversorgung(simulation_state, berichte, sensoren): """Erzeugt Demo-Daten fuer 400-V-Leistungskreis und 24-V-Steuerkreis.""" modus = simulation_state.get("anlagenmodus", "standard") sensor_warnungen = sum(1 for sensor in sensoren if sensor.get("status") != "OK") motor_rot = any(bericht.get("gesamtstatus") == "Rot" for bericht in berichte) if modus == "versorgungsfehler": status_24v = "Warnung" spannung_24v = 21.7 last_24v = 88 hinweis_24v = "24-V-Spannung niedrig. Sensorfehler koennen durch Unterspannung entstehen." elif sensor_warnungen >= 2: status_24v = "Warnung" spannung_24v = 23.1 last_24v = 74 hinweis_24v = "Mehrere Sensorsignale auffaellig. 24-V-Versorgung und 0V-Klemmen pruefen." else: status_24v = "OK" spannung_24v = round(random.uniform(24.0, 24.3), 1) last_24v = random.randint(54, 68) hinweis_24v = "24-V-Steuerkreis stabil." if modus == "400v_fehler": status_400v = "Fehler" werte_400v = (0, 0, 0) hinweis_400v = "400-V-Leistungskreis fehlt oder ist gesperrt. Versorgung, Sicherung und FU-Einspeisung pruefen." elif motor_rot and modus == "kritisch": status_400v = "OK" werte_400v = ( random.randint(398, 402), random.randint(398, 402), random.randint(398, 402), ) hinweis_400v = "400-V-Leistungskreis plausibel, Stoerung liegt eher an Antrieb/Mechanik." else: status_400v = "OK" werte_400v = ( random.randint(398, 402), random.randint(398, 402), random.randint(398, 402), ) hinweis_400v = "400-V-Leistungskreis stabil." gesamtstatus = "Fehler" if "Fehler" in [status_24v, status_400v] else "Warnung" if status_24v != "OK" or status_400v != "OK" else "OK" return { "hinweis": "Energieversorgung simuliert - keine echte Messung", "status": gesamtstatus, "versorgung_400v": { "l1_l2": werte_400v[0], "l2_l3": werte_400v[1], "l1_l3": werte_400v[2], "status": status_400v, "hinweis": hinweis_400v, }, "versorgung_24v": { "spannung": spannung_24v, "netzteil_last_prozent": last_24v, "status": status_24v, "hinweis": hinweis_24v, }, } def berechne_datenqualitaet(berichte, sensoren, sps_io, energieversorgung): """Bewertet Demo-Datenqualitaet und Diagnose-Sicherheit.""" gruende = [] score = 100 sensorfehler = [sensor for sensor in sensoren if sensor.get("status") != "OK"] sps_fehler = sps_io.get("diagnosen", []) plausibilitaet = [warnung for bericht in berichte for warnung in bericht.get("plausibilitaets_warnungen", [])] historienarm = [bericht for bericht in berichte if len(bericht.get("letzte_messungen", [])) < 3] if sensorfehler: score -= min(30, len(sensorfehler) * 10) gruende.append(f"{len(sensorfehler)} Sensor-/Signalauffaelligkeit(en)") if sps_fehler: score -= min(25, len(sps_fehler) * 8) gruende.append(f"{len(sps_fehler)} SPS-I/O-Diagnose(n)") if plausibilitaet: score -= min(20, len(plausibilitaet) * 6) gruende.append("Plausibilitaetswarnungen vorhanden") if energieversorgung.get("versorgung_24v", {}).get("status") != "OK": score -= 25 gruende.append("24-V-Steuerkreis nicht stabil") if energieversorgung.get("versorgung_400v", {}).get("status") != "OK": score -= 20 gruende.append("400-V-Leistungskreis auffaellig") if historienarm: score -= 5 gruende.append("wenig Historie fuer einzelne Motoren") sicherheit = max(0, min(100, score)) if sicherheit >= 85: status = "Gut" elif sicherheit >= 45: status = "Eingeschraenkt" else: status = "Kritisch" if energieversorgung.get("versorgung_400v", {}).get("status") != "OK" and sicherheit < 60: status = "Kritisch" return { "status": status, "diagnose_sicherheit_prozent": sicherheit, "gruende": gruende or ["Versorgung, Sensorwerte und SPS-I/O wirken plausibel"], "hinweis": "Demo-Bewertung der Datenqualitaet, keine zertifizierte Diagnose.", } def ist_sps_alarm_quittiert(sps_alarm, quittierungen): """Prueft, ob ein SPS-I/O-Alarm quittiert wurde.""" schluessel = f"sps:{sps_alarm['adresse']}" quittierung = quittierungen.get(schluessel) return ( quittierung is not None and quittierung.get("status") == sps_alarm["status"] ) def erstelle_sps_io(berichte, sensoren, quittierungen, simulation_state): """Simuliert SPS-Eingaenge, Ausgaenge und I/O-Diagnosen fuer Foerderband 1.""" motor = finde_bericht(berichte, "Motor Foerderband 1") or {} werte = {wert["messwert"]: wert for wert in motor.get("messwerte", [])} sensor_map = {sensor["id"]: sensor for sensor in sensoren} motor_ok = motor.get("gesamtstatus") != "Rot" motor_warnung = motor.get("gesamtstatus") == "Gelb" s1 = sensor_map.get("FB1-S1", {}) s2 = sensor_map.get("FB1-S2", {}) s3 = sensor_map.get("FB1-S3", {}) s1_wert = 0 if s1.get("status") == "Fehler" else int(str(s1.get("letzter_wert", "0")) == "1") s2_wert = 0 if s2.get("status") == "Fehler" else int(str(s2.get("letzter_wert", "0")) == "1") s3_wert = 0 if s3.get("status") == "Fehler" else int(str(s3.get("letzter_wert", "1")) == "1") modus = simulation_state.get("anlagenmodus") if modus == "normal": fu_stoerung = 0 motorschutz_ok = 1 elif modus in ["kritisch", "400v_fehler", "fu_stoerung"]: fu_stoerung = 1 motorschutz_ok = 1 else: fu_stoerung = 1 if motor.get("gesamtstatus") == "Rot" and random.random() < 0.35 else 0 motorschutz_ok = 0 if motor.get("gesamtstatus") == "Rot" and random.random() < 0.15 else 1 not_halt_ok = 1 tuer_ok = 1 start_aktiv = 0 if modus == "400v_fehler" else 1 if motorschutz_ok and not_halt_ok and tuer_ok else 0 drehzahl = werte.get("Drehzahl", {}).get("aktueller_wert", 0) if modus in ["fu_stoerung", "400v_fehler"]: drehzahl = 0 strom = werte.get("Stromaufnahme", {}).get("aktueller_wert", 0) temperatur = werte.get("Temperatur", {}).get("aktueller_wert", 0) vibration = werte.get("Vibration", {}).get("aktueller_wert", 0) eing_digital = [ io_eintrag("I0.0", "Not-Halt Kreis OK", "Digital", not_halt_ok, "OK", "Sicherheitskreis geschlossen", ["Not-Halt-Kreis und Sicherheitsrelais pruefen"]), io_eintrag("I0.1", "Motorschutzschalter OK", "Digital", motorschutz_ok, "OK" if motorschutz_ok else "Fehler", "Motorschutz OK" if motorschutz_ok else "Motorschutz ausgeloest", ["Ueberlast pruefen", "Motorstrom pruefen", "Mechanik auf Schwergaengigkeit pruefen"]), io_eintrag("I0.2", "Sensor S1 Materialeingang", "Digital", s1_wert, "OK" if s1_wert else "Fehler", "Material erkannt" if s1_wert else "Eingangssignal fehlt", ["Sensor S1 reinigen", "Kabel pruefen", "SPS-Eingang I0.2 pruefen"]), io_eintrag("I0.3", "Sensor S2 Materialausgang", "Digital", s2_wert, "OK" if s2_wert else "Fehler", "Material am Ausgang erkannt" if s2_wert else "Ausgangssignal fehlt", ["Sensor S2 pruefen", "Materialstau pruefen", "SPS-Eingang I0.3 pruefen"]), io_eintrag("I0.4", "Sensor S3 Bandlaufueberwachung", "Digital", s3_wert, "OK" if s3_wert else "Fehler", "Bandlauf OK" if s3_wert else "Bandlauf pruefen oder Sensor/Kabel pruefen", ["Bandlauf pruefen", "Sensor S3 und Verdrahtung pruefen"]), io_eintrag("I0.5", "Drehzahlimpuls Motor M1", "Digital", 1 if drehzahl > 0 else 0, "OK" if drehzahl > 0 else "Fehler", "Drehzahlimpuls vorhanden" if drehzahl > 0 else "Drehzahlimpuls fehlt", ["Drehzahlsensor", "Klemme X15", "Digitaleingang I0.5 pruefen"]), io_eintrag("I0.6", "Freigabe Sicherheitstuer", "Digital", tuer_ok, "OK", "Sicherheitstuer freigegeben", ["Tuerkontakt und Sicherheitskreis pruefen"]), io_eintrag("I0.7", "Stoerung Frequenzumrichter", "Digital", fu_stoerung, "Fehler" if fu_stoerung else "OK", "FU-Stoerung aktiv" if fu_stoerung else "FU meldet keine Stoerung", ["FU-Fehlerspeicher pruefen", "Reset nur nach Ursachenpruefung", "Motor und Last pruefen"]), ] eing_analog = [ io_eintrag("IW64", "Stromaufnahme Motor M1", "Analog", strom, werte.get("Stromaufnahme", {}).get("status", "OK").replace("Gruen", "OK").replace("Gelb", "Warnung").replace("Rot", "Fehler"), "Stromaufnahme plausibel", ["Messwandler und Analogeingang pruefen"], "A"), io_eintrag("IW66", "Temperatur Motor M1", "Analog", temperatur, werte.get("Temperatur", {}).get("status", "OK").replace("Gruen", "OK").replace("Gelb", "Warnung").replace("Rot", "Fehler"), "Temperaturwert aus Motorfuehler", ["PT100, Leitung und Analogeingang pruefen"], "°C"), io_eintrag("IW68", "Vibration Motor M1", "Analog", vibration, werte.get("Vibration", {}).get("status", "OK").replace("Gruen", "OK").replace("Gelb", "Warnung").replace("Rot", "Fehler"), "Vibrationswert aus Sensor", ["Vibrationssensor und Befestigung pruefen"], "mm/s"), io_eintrag("IW70", "Drehzahl Istwert", "Analog", drehzahl, werte.get("Drehzahl", {}).get("status", "OK").replace("Gruen", "OK").replace("Gelb", "Warnung").replace("Rot", "Fehler"), "Drehzahl-Istwert plausibel" if drehzahl > 0 else "Drehzahl fehlt", ["Drehzahlsensor und Skalierung pruefen"], "U/min"), ] ausg_digital = [ io_eintrag("Q0.0", "Motor M1 Freigabe", "Digital", 1 if start_aktiv else 0, "aktiv" if start_aktiv else "inaktiv", "Motorfreigabe aktiv" if start_aktiv else "Motorfreigabe gesperrt", ["SPS-Ausgang Q0.0 und Verdrahtung pruefen"]), io_eintrag("Q0.1", "Frequenzumrichter Start", "Digital", 1 if start_aktiv else 0, "aktiv" if start_aktiv else "inaktiv", "Startsignal an FU aktiv" if start_aktiv else "Startsignal nicht aktiv", ["SPS-Ausgang Q0.1", "FU-Startklemme pruefen"]), io_eintrag("Q0.2", "Warnleuchte Gelb", "Digital", 1 if motor_warnung else 0, "aktiv" if motor_warnung else "inaktiv", "Warnung liegt an" if motor_warnung else "Keine Warnleuchte aktiv"), io_eintrag("Q0.3", "Stoerleuchte Rot", "Digital", 1 if not motor_ok else 0, "aktiv" if not motor_ok else "inaktiv", "Kritische Stoerung liegt an" if not motor_ok else "Keine Stoerleuchte aktiv"), io_eintrag("Q0.4", "Hupe / akustischer Alarm", "Digital", 1 if not motor_ok and not motor.get("alarm_quittiert") else 0, "aktiv" if not motor_ok and not motor.get("alarm_quittiert") else "inaktiv", "Akustischer Alarm aktiv" if not motor_ok and not motor.get("alarm_quittiert") else "Hupe aus"), io_eintrag("Q0.5", "Reset Frequenzumrichter", "Digital", 0, "inaktiv", "Reset nicht gesetzt", ["Reset nur nach Ursachenpruefung betaetigen"]), ] ausg_analog = [ io_eintrag("QW64", "FU Sollwert Drehzahl", "Analog", motor.get("maschinenakte", {}).get("sollwert_drehzahl", 1450) if start_aktiv else 0, "aktiv" if start_aktiv else "inaktiv", "Sollwert plausibel" if start_aktiv else "Sollwert 0 wegen fehlender Freigabe", ["Sollwertkette und FU-Parameter pruefen"], "U/min"), io_eintrag("QW66", "Drehmomentbegrenzung / Lastvorgabe Demo", "Analog", 80 if motor_ok else 50, "aktiv", "Demo-Lastvorgabe fuer Antrieb", ["FU-Parameter und Drehmomentbegrenzung pruefen"], "%"), ] diagnosen = [] def add_diagnose(adresse, kategorie, diagnose, pruefungen, prioritaet): eintrag = { "adresse": adresse, "kategorie": kategorie, "status": "Rot" if prioritaet == "Hoch" else "Gelb", "diagnose": diagnose, "empfohlene_pruefung": pruefungen, "prioritaet": prioritaet, "alarm_quittiert": False, } eintrag["alarm_quittiert"] = ist_sps_alarm_quittiert(eintrag, quittierungen) diagnosen.append(eintrag) if start_aktiv and drehzahl == 0: add_diagnose("Q0.1", "SPS Ausgang / Motor", "SPS gibt Startsignal aus, aber Motor laeuft nicht.", ["FU bereit pruefen", "Motorschutz pruefen", "Motor blockiert?", "Drehzahlsensor pruefen"], "Hoch") if fu_stoerung: add_diagnose("I0.7", "Frequenzumrichter", "Frequenzumrichter meldet Stoerung.", ["FU-Fehlerspeicher pruefen", "Reset nur nach Ursachenpruefung", "Motor und Last pruefen"], "Hoch") if not motorschutz_ok: add_diagnose("I0.1", "Motorschutz", "Motorschutz ausgeloest.", ["Ueberlast pruefen", "Motorstrom pruefen", "Mechanik pruefen", "Motorschutz erst nach Pruefung zuruecksetzen"], "Hoch") if not s1_wert and modus == "sensorfehler": add_diagnose("I0.2", "Sensor Eingang", "Sensor S1 zeigt trotz erwartetem Material kein Eingangssignal.", ["Sensor reinigen", "Sensor defekt?", "Kabelbruch?", "SPS-Eingang I0.2 pruefen"], "Mittel") if not s2_wert and modus in ["materialstau", "sensorfehler"]: add_diagnose("I0.3", "Sensor Ausgang", "Sensor S2 zeigt im Stoerbild kein Ausgangssignal.", ["Materialstau pruefen", "Sensor reinigen", "SPS-Eingang I0.3 pruefen"], "Mittel") if not s3_wert: add_diagnose("I0.4", "Bandlauf", "Bandlaufueberwachung meldet Fehler.", ["Bandlauf pruefen", "Sensor/Kabel pruefen", "Mechanische Fuehrung kontrollieren"], "Mittel") if not motor_ok: add_diagnose("Q0.3", "Stoermeldung", "Stoerleuchte Rot aktiv: mindestens ein kritischer Alarm liegt an.", ["Aktive Alarme pruefen", "Ursache beheben", "Alarm quittieren"], "Hoch") if modus == "versorgungsfehler": add_diagnose("24V", "Steuerspannung", "24-V-Steuerspannung niedrig, Sensorsignale koennen unplausibel werden.", ["24-V-Netzteil messen", "0V-Klemmen pruefen", "Sensorversorgung unter Last pruefen"], "Mittel") if modus == "400v_fehler": add_diagnose("400V", "Leistungskreis", "400-V-Leistungskreis nicht verfuegbar, Motorstart gesperrt.", ["Hauptschalter und Sicherungen pruefen", "FU-Einspeisung messen", "Freigabekette pruefen"], "Hoch") return { "hinweis": "SPS I/O-Daten simuliert - keine echte SPS-Verbindung", "eingänge_digital": eing_digital, "eingänge_analog": eing_analog, "ausgänge_digital": ausg_digital, "ausgänge_analog": ausg_analog, "diagnosen": diagnosen, } def aktualisiere_sps_quittierungen(sps_io, quittierungen): """Setzt SPS-Quittierungen zurueck, wenn die Diagnose verschwunden ist.""" aktive_schluessel = {f"sps:{diagnose['adresse']}" for diagnose in sps_io.get("diagnosen", [])} for schluessel in list(quittierungen): if schluessel.startswith("sps:") and schluessel not in aktive_schluessel: del quittierungen[schluessel] speichere_quittierungen(quittierungen) return quittierungen def erstelle_sps_alarme(sps_io): """Wandelt SPS-I/O-Diagnosen in quittierbare Dashboard-Alarme um.""" alarme = [] for diagnose in sps_io.get("diagnosen", []): alarme.append({ "typ": "sps", "id": diagnose["adresse"], "motorname": f"SPS {diagnose['adresse']}", "bereich": "SPS I/O Foerderband 1", "status": diagnose["status"], "meldungstyp": "SPS-I/O-Diagnose", "risikowert": 75 if diagnose["status"] == "Rot" else 45, "meldung": diagnose["diagnose"], "alarm_quittiert": diagnose["alarm_quittiert"], }) return alarme def erstelle_schrittkette(berichte, sensoren, sps_io, simulation_state, energieversorgung=None): """Erstellt eine simulierte Ablaufdiagnose fuer Foerderband 1.""" motor = finde_bericht(berichte, "Motor Foerderband 1") or {} sensoren_status = {sensor["id"]: sensor["status"] for sensor in sensoren} digital = {eintrag["adresse"]: eintrag for eintrag in sps_io.get("eingänge_digital", [])} ausgang = {eintrag["adresse"]: eintrag for eintrag in sps_io.get("ausgänge_digital", [])} analog = {eintrag["adresse"]: eintrag for eintrag in sps_io.get("eingänge_analog", [])} energieversorgung = energieversorgung or {} modus = simulation_state.get("anlagenmodus", "standard") betriebsart = simulation_state.get("betriebsart", "automatik") stoerung = False aktiver_status = "aktiv" jetzt = datetime.now() schrittzeit_s = 4 zyklus_schritte = list(range(4, 12)) if modus in ["normal", "meisterfreigabe"] and betriebsart == "automatik" else list(range(0, 12)) aktueller_schritt = zyklus_schritte[int(jetzt.timestamp() // schrittzeit_s) % len(zyklus_schritte)] aktiv_seit_s = int(jetzt.timestamp()) % schrittzeit_s max_schrittzeit_s = 8 diagnose = "Automatik aktiv - Schrittfolge laeuft simuliert weiter." versorgung_400v_ok = energieversorgung.get("versorgung_400v", {}).get("status", "OK") == "OK" versorgung_24v_ok = energieversorgung.get("versorgung_24v", {}).get("status", "OK") == "OK" fu_ok = digital.get("I0.7", {}).get("wert", 0) == 0 startfreigabe = [ {"name": "Not-Halt Kreis", "adresse": "I0.0", "ok": digital.get("I0.0", {}).get("wert", 1) == 1}, {"name": "Sicherheit / Schutzkreis", "adresse": "I0.6", "ok": digital.get("I0.6", {}).get("wert", 1) == 1}, {"name": "Motorschutz", "adresse": "I0.1", "ok": digital.get("I0.1", {}).get("wert", 1) == 1}, {"name": "FU bereit", "adresse": "I0.7", "ok": fu_ok}, {"name": "400 V Versorgung", "adresse": "400V", "ok": versorgung_400v_ok}, {"name": "24 V Versorgung", "adresse": "24V", "ok": versorgung_24v_ok}, {"name": "Automatikbetrieb aktiv", "adresse": "Betriebsart", "ok": betriebsart == "automatik"}, ] fehlende_freigaben = [eintrag for eintrag in startfreigabe if not eintrag["ok"]] if betriebsart == "handbetrieb": aktueller_schritt = 1 aktiver_status = "wartet" diagnose = "Handbetrieb aktiv - automatische Schrittfolge pausiert, Materialfluss steht." elif betriebsart == "wartung": aktueller_schritt = 0 aktiver_status = "wartet" diagnose = "Wartungsbetrieb aktiv - Startfreigabe gesperrt, Meisterfreigabe fuer Wiederanlauf erforderlich." elif betriebsart == "gestoppt": aktueller_schritt = 0 aktiver_status = "wartet" diagnose = "Anlage gestoppt - Materialfluss steht, keine automatische Schrittfolge aktiv." elif betriebsart == "stoerung": aktueller_schritt = 99 stoerung = True diagnose = "Demo-Betriebsart Stoerung aktiv - Automatik gestoppt und Quittierung erforderlich." elif modus in ["normal", "reparatur_demo", "meisterfreigabe"]: diagnose = "Normalbetrieb aktiv - Schrittfolge laeuft simuliert weiter." elif modus == "pruefprotokoll_erforderlich": aktueller_schritt = 0 aktiver_status = "wartet" diagnose = "Pruefprotokoll erforderlich - Wiederanlauf erst nach Pruefung und Freigabe." elif modus == "materialstau": aktueller_schritt = 6 aktiver_status = "warnung" aktiv_seit_s = 11 diagnose = "Schrittzeit ueberschritten: Materialstau zwischen S1 und S2 moeglich." elif modus == "fu_stoerung": aktueller_schritt = 5 aktiver_status = "warnung" aktiv_seit_s = 9 diagnose = "M1 Start: Startsignal aktiv, aber FU-Stoerung/Drehzahlrueckmeldung fehlt." elif motor.get("gesamtstatus") == "Rot": aktueller_schritt = 99 stoerung = True diagnose = "Automatik gestoppt wegen kritischem Motorstatus." elif digital.get("I0.7", {}).get("wert") == 1: aktueller_schritt = 99 stoerung = True diagnose = "Ablauf in Stoerung. FU-Fehlerspeicher pruefen." elif digital.get("I0.0", {}).get("wert") == 0: aktueller_schritt = 1 diagnose = "Startfreigabe fehlt: Not-Halt-Kreis nicht OK." elif digital.get("I0.1", {}).get("wert") == 0: aktueller_schritt = 2 diagnose = "Schritt 2 aktiv, aber I0.1 Motorschutz OK = 0. Motorstart gesperrt." elif sensoren_status.get("FB1-S1") == "Fehler": aktueller_schritt = 3 diagnose = "Warte auf Materialeingang. Sensor S1 oder Materialzufuehrung pruefen." elif sensoren_status.get("FB1-S2") == "Fehler": aktueller_schritt = 5 diagnose = "Material wurde nicht am Ausgang erkannt. Sensor S2, Bandlauf oder Materialstau pruefen." elif ausgang.get("Q0.1", {}).get("wert") == 1 and analog.get("IW70", {}).get("wert") == 0: aktueller_schritt = 5 diagnose = "Startsignal aktiv, aber keine Drehzahlrueckmeldung. FU, Motor, Motorschutz oder Drehzahlsensor pruefen." elif motor.get("gesamtstatus") == "Gelb": diagnose = "Ablauf verlangsamt / Werte beobachten." schritt_infos = [ (0, "Grundstellung", "Anlage bereit", "Keine Stoerung aktiv", "grundstellung"), (1, "Startfreigabe", "Freigaben pruefen", "Not-Halt, Schutzkreis, Motorschutz, Automatik", "startfreigabe"), (2, "Versorgung", "400 V / 24 V pruefen", "400 V OK und 24 V OK", "versorgung"), (3, "SPS/FU bereit", "SPS/FU-Status auswerten", "I0.7 FU-Stoerung = 0", "sps_fu"), (4, "S1 Eingang", "Materialeingang auswerten", "I0.2 S1 = 1", "s1"), (5, "M1 Start", "Q0.0 Motorfreigabe und Q0.1 FU Start aktiv", "Drehzahl > 0", "m1"), (6, "Foerdern", "Foerderband laeuft", "Material erreicht S2 innerhalb Schrittzeit", "band"), (7, "S2 Ausgang", "S2 Materialausgang auswerten", "I0.3 S2 = 1", "s2"), (8, "Pruefstation bereit", "Pruefstation freigeben", "Station bereit", "pruefung"), (9, "Teil pruefen", "Pruefung simuliert aktiv", "Pruefergebnis OK", "pruefung"), (10, "Ausschleusen", "Teil freigeben / ausschleusen", "Ausgang frei", "ausgang"), (11, "Uebergabe fertig", "Zaehler erhoehen", "Material uebergeben", "ausgang"), (99, "Stoerung / Freigabe fehlt", "Ablauf stoppen", "Stoerung oder Freigabe fehlt", "stoerung"), ] schritte = [] for nummer, name, aktion, bedingung, komponente in schritt_infos: if nummer == 99: status = "aktiv" if aktueller_schritt == 99 else "inaktiv" elif stoerung: status = "wartet" elif nummer < aktueller_schritt: status = "erledigt" elif nummer == aktueller_schritt: status = aktiver_status else: status = "wartet" schritte.append({ "nummer": nummer, "name": name, "status": status, "aktion": aktion, "bedingung": bedingung, "komponente": komponente, "aktiv_seit_text": f"00:{aktiv_seit_s:02d}" if nummer == aktueller_schritt else "00:00", "max_schrittzeit_text": f"00:{max_schrittzeit_s:02d}", "schrittzeit_status": "ueberschritten" if nummer == aktueller_schritt and aktiv_seit_s > max_schrittzeit_s else "OK", }) aktueller_name = next(s["name"] for s in schritte if s["nummer"] == aktueller_schritt) startfreigabe_status = "OK" if not fehlende_freigaben else "gesperrt" materialfluss = "aktiv" if betriebsart == "automatik" and aktueller_schritt in [5, 6, 7, 8, 9, 10] and not stoerung and modus not in ["materialstau", "fu_stoerung"] else "steht" return { "hinweis": "Schrittkette simuliert - keine echte SPS-/GRAFCET-Verbindung", "name": "Linie 1 Ablauf", "aktueller_schritt": aktueller_schritt, "aktueller_schritt_name": aktueller_name, "status": "Stoerung" if stoerung else "Pausiert" if betriebsart in ["handbetrieb", "wartung", "gestoppt"] else "Aktiv", "stoerung": stoerung, "ablaufdiagnose": diagnose, "betriebsart": betriebsart, "startfreigabe_status": startfreigabe_status, "fehlende_freigaben": [eintrag["name"] for eintrag in fehlende_freigaben], "startfreigabe": startfreigabe, "materialfluss": materialfluss, "wiederanlauf": "Meisterfreigabe erforderlich" if betriebsart == "wartung" or stoerung else "freigegeben", "aktiv_seit_text": f"00:{aktiv_seit_s:02d}", "max_schrittzeit_text": f"00:{max_schrittzeit_s:02d}", "schrittzeit_status": "ueberschritten" if aktiv_seit_s > max_schrittzeit_s else "OK", "schritte": schritte, } def lese_vorherige_dashboard_daten(): """Liest die vorherigen Dashboard-Daten fuer Zustandsvergleiche.""" if not os.path.exists(DATEN_PFAD): return {} try: with open(DATEN_PFAD, mode="r", encoding="utf-8") as datei: return json.load(datei) except json.JSONDecodeError: return {} def speichere_ereignisse_fuer_messrunde(berichte, vorherige_daten): """Speichert wichtige neue Ereignisse aus einer Messrunde.""" vorherige_motoren = { motor["motorname"]: motor for motor in vorherige_daten.get("motoren", []) } for bericht in berichte: vorheriger_motor = vorherige_motoren.get(bericht["motorname"], {}) vorheriger_status = vorheriger_motor.get("gesamtstatus") vorheriges_risiko = vorheriger_motor.get("risikowert", 0) if bericht["gesamtstatus"] != "Gruen" and vorheriger_status != bericht["gesamtstatus"]: speichere_ereignis( bericht["motorname"], bericht["bereich"], bericht["gesamtstatus"], bericht["meldungstyp"], bericht["risikowert"], f"{bericht['motorname']} wurde {bericht['gesamtstatus']}", bericht["alarm_quittiert"], ) speichere_ereignis( bericht["motorname"], bericht["bereich"], bericht["gesamtstatus"], bericht["meldungstyp"], bericht["risikowert"], "Wartungsempfehlung erstellt", bericht["alarm_quittiert"], ) if bericht["risikowert"] >= 60 and bericht["risikowert"] - vorheriges_risiko >= 10: speichere_ereignis( bericht["motorname"], bericht["bereich"], bericht["gesamtstatus"], bericht["meldungstyp"], bericht["risikowert"], f"Risiko auf {bericht['risikowert']} erhoeht", bericht["alarm_quittiert"], ) vorherige_messwerte = { wert["messwert"]: wert for wert in vorheriger_motor.get("messwerte", []) } for wert in bericht["messwerte"]: vorheriger_wert = vorherige_messwerte.get(wert["messwert"], {}) if wert["status"] != "Gruen" and vorheriger_wert.get("status") != wert["status"]: speichere_ereignis( bericht["motorname"], bericht["bereich"], bericht["gesamtstatus"], bericht["meldungstyp"], bericht["risikowert"], f"{wert['messwert']} ueberschreitet Grenzwert", bericht["alarm_quittiert"], ) def berechne_langzeittrend(motorname, aktuelle_durchschnittsabweichung, historie): """Bewertet den Durchschnitt der letzten 5 Abweichungen.""" werte = historie.get(motorname, []) + [aktuelle_durchschnittsabweichung] letzte_werte = werte[-5:] if len(letzte_werte) < 5: return "Noch zu wenig Langzeitdaten" durchschnitt = sum(letzte_werte) / len(letzte_werte) if durchschnitt > 25: return "Langzeittrend kritisch" if durchschnitt > 10: return "Langzeittrend auffaellig" return "Langzeittrend unauffaellig" def ermittle_letzte_messungen(motorname, aktuelle_durchschnittsabweichung, historie): """Gibt die letzten 5 Durchschnittsabweichungen fuer das Dashboard zurueck.""" werte = historie.get(motorname, []) + [aktuelle_durchschnittsabweichung] return [round(wert, 1) for wert in werte[-5:]] def werte_motor_aus(motor, aktuelle_werte, historie, simulation_state): """Vergleicht aktuelle Werte mit Normalwerten und erstellt einen Bericht.""" ergebnisse = [] warnmeldungen = [] abweichungen = [] for messwert_name, normalwert in motor["normalwerte"].items(): aktueller_wert = aktuelle_werte[messwert_name] abweichung = berechne_abweichung_prozent(normalwert, aktueller_wert) if messwert_name == "Temperatur": status = bewerte_temperatur(aktueller_wert) else: status = bewerte_abweichung(abweichung) warnmeldung = ermittle_warnmeldung(messwert_name, status, abweichung) if status != "Gruen": warnmeldungen.append(warnmeldung) abweichungen.append(abs(abweichung)) ergebnisse.append( { "messwert": messwert_name, "normalwert": round(normalwert, 2), "aktueller_wert": aktueller_wert, "einheit": EINHEITEN[messwert_name], "abweichung_prozent": round(abweichung, 1), "status": status, "warnmeldung": warnmeldung, } ) status_liste = [ergebnis["status"] for ergebnis in ergebnisse] gesamtstatus = ermittle_gesamtstatus(status_liste) durchschnittsabweichung = sum(abweichungen) / len(abweichungen) kombinationsdiagnosen = ermittle_kombinationsdiagnosen(ergebnisse) langzeittrend = berechne_langzeittrend( motor["name"], durchschnittsabweichung, historie, ) letzte_messungen = ermittle_letzte_messungen( motor["name"], durchschnittsabweichung, historie, ) moegliche_ursachen = kombinationsdiagnosen + warnmeldungen risikowert = berechne_risikowert( gesamtstatus, durchschnittsabweichung, langzeittrend, kombinationsdiagnosen, motor["produktionsrelevanz"], ) ausfallprognose_text, geschaetzte_tage = berechne_ausfallprognose(risikowert) wartungsempfehlung = erstelle_wartungsempfehlung(gesamtstatus, risikowert) empfohlenes_wartungsfenster = waehle_wartungsfenster(geschaetzte_tage) empfohlene_massnahmen = ermittle_empfohlene_massnahmen(ergebnisse) ersatzteile_pruefbereiche = ermittle_ersatzteile(ergebnisse) wirtschaftliche_bewertung = berechne_wirtschaftliche_bewertung(motor) energie_mehrkosten = berechne_energie_mehrkosten(motor, ergebnisse) wartungsstatus = berechne_wartungsstatus(motor) plausibilitaets_warnungen = pruefe_plausibilitaet(motor, aktuelle_werte, historie) elektronische_fehler = diagnostiziere_elektronikfehler(motor, aktuelle_werte, historie) fehlerwahrscheinlichkeiten = berechne_fehlerwahrscheinlichkeiten(ergebnisse) meldungstyp = ermittle_meldungstyp(gesamtstatus) instandhaltungsprioritaet = ermittle_prioritaet(risikowert) trendrichtung = ermittle_trendrichtung(letzte_messungen) aktuelle_fehlerbilder = ermittle_aktuelle_fehlerbilder(ergebnisse, elektronische_fehler) reparatur_hinweise = analysiere_reparaturhistorie(motor["name"], aktuelle_fehlerbilder) nicht_nochmal_tauschen_warnungen = ermittle_nicht_nochmal_tauschen_warnungen( motor["name"], aktuelle_fehlerbilder, ) diagnose_trennung = trenne_mechanische_elektrische_diagnose( ergebnisse, kombinationsdiagnosen, elektronische_fehler, ) verbaute_bauteile = markiere_betroffene_bauteile( VERBAUTE_BAUTEILE[motor["name"]], ergebnisse, elektronische_fehler, ) betriebsempfehlung = erstelle_betriebsempfehlung({ "gesamtstatus": gesamtstatus, "risikowert": risikowert, }) return { "motorname": motor["name"], "bereich": motor["bereich"], "produktionsrelevanz": motor["produktionsrelevanz"], "produktionsauswirkung": PRODUKTIONS_AUSWIRKUNGEN.get(motor["name"], { "relevanz": "Niedrig", "auswirkung": "Geringe Auswirkung auf die Produktion", "teileverlust_pro_stunde": 0, "empfehlung": "Routinepruefung ausreichend", }), "maschinenakte": motor["maschinenakte"], "sps_zuordnung": SPS_ZUORDNUNG_FOERDERBAND_1 if motor["name"] == "Motor Foerderband 1" else {}, "betriebsdaten": wartungsstatus, "simulierter_zustand": ermittle_simulierten_motorzustand(motor, simulation_state), "zeitstempel": aktuelle_werte["Zeitstempel"], "gesamtstatus": gesamtstatus, "meldungstyp": meldungstyp, "instandhaltungsprioritaet": instandhaltungsprioritaet, "durchschnittsabweichung": round(durchschnittsabweichung, 1), "langzeittrend": langzeittrend, "letzte_messungen": letzte_messungen, "trendrichtung": trendrichtung, "plausibilitaets_warnungen": plausibilitaets_warnungen, "elektronische_fehler": elektronische_fehler, "aktuelle_fehlerbilder": aktuelle_fehlerbilder, "reparatur_hinweise": reparatur_hinweise, "nicht_nochmal_tauschen_warnungen": nicht_nochmal_tauschen_warnungen, "diagnose_trennung": diagnose_trennung, "empfohlenes_gewerk": diagnose_trennung["empfohlenes_gewerk"], "betriebsempfehlung": betriebsempfehlung, "moegliche_ursachen": moegliche_ursachen or ["Keine Auffaelligkeit"], "kombinationsdiagnosen": kombinationsdiagnosen, "risikowert": risikowert, "ausfallprognose_text": ausfallprognose_text, "geschaetzte_tage_bis_ausfall": geschaetzte_tage, "wartungsempfehlung": wartungsempfehlung, "empfohlenes_wartungsfenster": empfohlenes_wartungsfenster, "empfohlene_massnahmen": empfohlene_massnahmen, "ersatzteile_pruefbereiche": ersatzteile_pruefbereiche, "fehlerwahrscheinlichkeiten": fehlerwahrscheinlichkeiten, "wirtschaftliche_bewertung": wirtschaftliche_bewertung, "energie_mehrkosten": energie_mehrkosten, "verbaute_bauteile": verbaute_bauteile, "wartungsbericht_erforderlich": gesamtstatus == "Rot" or risikowert > 60, "alarm_quittiert": False, "messwerte": ergebnisse, } def speichere_historie(berichte): """Speichert jede Messrunde in motor_history.csv.""" datei_existiert = os.path.exists(HISTORIE_PFAD) with open(HISTORIE_PFAD, mode="a", encoding="utf-8", newline="") as datei: writer = csv.DictWriter(datei, fieldnames=HISTORIE_FELDNAMEN) if not datei_existiert: writer.writeheader() for bericht in berichte: werte = { ergebnis["messwert"]: ergebnis["aktueller_wert"] for ergebnis in bericht["messwerte"] } writer.writerow( { "Zeitstempel": bericht["zeitstempel"], "Motorname": bericht["motorname"], "Gesamtstatus": bericht["gesamtstatus"], "Stromaufnahme": werte["Stromaufnahme"], "Temperatur": werte["Temperatur"], "Vibration": werte["Vibration"], "Drehzahl": werte["Drehzahl"], "DurchschnittAbweichung": bericht["durchschnittsabweichung"], } ) begrenze_csv_zeilen(HISTORIE_PFAD, MAX_HISTORIE_ZEILEN) def begrenze_csv_zeilen(pfad, max_datenzeilen): """Begrenzt CSV-Dateien robust auf die letzten Datenzeilen und behaelt den Header.""" if max_datenzeilen <= 0 or not os.path.exists(pfad): return try: with open(pfad, mode="r", encoding="utf-8", newline="") as datei: zeilen = datei.readlines() except OSError: return if len(zeilen) <= max_datenzeilen + 1: return header = zeilen[:1] daten = zeilen[1:][-max_datenzeilen:] temporaerer_pfad = f"{pfad}.tmp" try: with open(temporaerer_pfad, mode="w", encoding="utf-8", newline="") as datei: datei.writelines(header + daten) os.replace(temporaerer_pfad, pfad) except OSError: if os.path.exists(temporaerer_pfad): os.remove(temporaerer_pfad) def erstelle_status_zaehler(berichte): """Zaehlt Motoren je Gesamtstatus fuer den Dashboard-Kopfbereich.""" zaehler = { "gesamt": len(berichte), "gruen": 0, "gelb": 0, "rot": 0, } for bericht in berichte: if bericht["gesamtstatus"] == "Gruen": zaehler["gruen"] += 1 elif bericht["gesamtstatus"] == "Gelb": zaehler["gelb"] += 1 elif bericht["gesamtstatus"] == "Rot": zaehler["rot"] += 1 return zaehler def erstelle_alarme(berichte): """Erstellt eine kurze Liste aktiver gelber und roter Alarme.""" alarme = [] for bericht in berichte: if bericht["gesamtstatus"] == "Gruen": continue alarme.append( { "typ": "motor", "id": bericht["motorname"], "motorname": bericht["motorname"], "bereich": bericht["bereich"], "status": bericht["gesamtstatus"], "meldungstyp": bericht["meldungstyp"], "risikowert": bericht["risikowert"], "meldung": bericht["moegliche_ursachen"][0], "alarm_quittiert": bericht["alarm_quittiert"], } ) return alarme def fehlercode_kurz(fehlercode): """Gibt eine kompakte Fehlercode-Info fuer Alarme und Motoren zurueck.""" eintrag = next((code for code in FEHLERCODE_KATALOG if code["fehlercode"] == fehlercode), None) if not eintrag: return None return { "fehlercode": eintrag["fehlercode"], "meldung": eintrag["meldung"], "kategorie": eintrag["kategorie"], "prioritaet": eintrag["prioritaet"], "gewerk": eintrag["gewerk"], "kurzempfehlung": eintrag["kurzempfehlung"], } def fehlercodes_fuer_motor(bericht): """Ordnet einem Motor passende Fehlercodes aus dem Demo-Katalog zu.""" codes = [] werte = {eintrag["messwert"]: eintrag for eintrag in bericht.get("messwerte", [])} if werte.get("Vibration", {}).get("status") in ["Gelb", "Rot"]: codes.append("FB1-ERR-002") temperatur = werte.get("Temperatur", {}).get("aktueller_wert", 0) if temperatur <= -20 or any("Temperatur" in warnung for warnung in bericht.get("plausibilitaets_warnungen", [])): codes.append("FB1-ERR-003") if any("Frequenzumrichter" in fehler.get("kategorie", "") for fehler in bericht.get("elektronische_fehler", [])): codes.append("FB1-ERR-004") return [fehlercode_kurz(code) for code in dict.fromkeys(codes) if fehlercode_kurz(code)] def fehlercodes_fuer_sensor(sensor): """Ordnet Sensoralarmen passende Fehlercodes zu.""" if sensor.get("status") == "OK": return [] if sensor.get("meldungstyp") == "Materialflussalarm": return [fehlercode_kurz("FB1-ERR-008")] if sensor.get("id") in ["FB1-S1", "FB1-S2"]: return [fehlercode_kurz("FB1-ERR-005")] if sensor.get("id") == "FB1-S3": return [fehlercode_kurz("FB1-ERR-007")] return [] def fehlercodes_fuer_sps_diagnose(diagnose): """Ordnet SPS-I/O-Diagnosen passende Fehlercodes zu.""" adresse = diagnose.get("adresse") text = diagnose.get("diagnose", "") codes = [] if adresse == "Q0.1" or "Startsignal" in text: codes.append("FB1-ERR-001") if adresse == "I0.7" or "FU" in text: codes.append("FB1-ERR-004") if adresse == "I0.1" or "Motorschutz" in text: codes.append("FB1-ERR-006") if adresse == "I0.4" or "Bandlauf" in text: codes.append("FB1-ERR-007") if adresse == "I0.3" and "Materialstau" in text: codes.append("FB1-ERR-008") return [fehlercode_kurz(code) for code in dict.fromkeys(codes) if fehlercode_kurz(code)] def ergaenze_passende_fehlercodes(berichte, sensoren, sps_io, schrittkette, aktive_alarme): """Ergaenzt Motoren und aktive Alarme um passende Fehlercodes.""" motor_codes = {} sensor_codes = {} sps_codes = {} for bericht in berichte: codes = fehlercodes_fuer_motor(bericht) bericht["passende_fehlercodes"] = codes motor_codes[bericht["motorname"]] = codes for sensor in sensoren: codes = fehlercodes_fuer_sensor(sensor) sensor["passende_fehlercodes"] = codes sensor_codes[sensor["id"]] = codes for diagnose in sps_io.get("diagnosen", []): codes = fehlercodes_fuer_sps_diagnose(diagnose) diagnose["passende_fehlercodes"] = codes sps_codes[diagnose["adresse"]] = codes if schrittkette.get("aktueller_schritt") == 2: sps_codes.setdefault("Q0.1", []) code = fehlercode_kurz("FB1-ERR-001") if code and code not in sps_codes["Q0.1"]: sps_codes["Q0.1"].append(code) for alarm in aktive_alarme: if alarm.get("typ") == "sensor": alarm["passende_fehlercodes"] = sensor_codes.get(alarm.get("id"), []) elif alarm.get("typ") == "sps": alarm["passende_fehlercodes"] = sps_codes.get(alarm.get("id"), []) else: alarm["passende_fehlercodes"] = motor_codes.get(alarm.get("motorname"), []) def alarm_schluessel(alarm): """Erstellt einen stabilen Schluessel fuer den Alarm-Timer.""" typ = alarm.get("typ", "motor") if typ == "motor": return f"motor:{alarm.get('id') or alarm.get('motorname')}" return f"{typ}:{alarm.get('id') or alarm.get('motorname')}" def formatiere_alarmdauer(sekunden): """Formatiert Sekunden als kurze Alarmdauer.""" sekunden = max(int(sekunden), 0) stunden = sekunden // 3600 minuten = (sekunden % 3600) // 60 rest = sekunden % 60 if stunden: return f"{stunden} h {minuten:02d} min" if minuten: return f"{minuten} min {rest:02d} s" return f"00:{rest:02d} s" def eskalation_fuer_alarm(alarm, sekunden): """Erstellt eine einfache Demo-Eskalation aus Alarmdauer und Status.""" hinweise = [] if sekunden >= 30 * 60: status = "Meister informieren" hinweise.append("Alarm laenger als 30 Minuten aktiv - Eskalation empfohlen") elif sekunden >= 10 * 60: status = "Instandhaltung informieren" hinweise.append("Alarm laenger als 10 Minuten aktiv - Instandhaltung pruefen") elif sekunden >= 5 * 60: status = "Beobachten" hinweise.append("Alarm laenger als 5 Minuten aktiv") else: status = "Neu" if alarm.get("status") == "Rot" and not alarm.get("alarm_quittiert"): hinweise.append("Kritischer Alarm nicht quittiert") if alarm.get("status") == "Rot": hinweise.append("Produktionsverlust laeuft moeglicherweise") return status, hinweise def aktualisiere_alarmdauer(aktive_alarme): """Aktualisiert Startzeit, Dauer und Eskalation fuer aktive Alarme.""" state = lade_active_alarm_state() jetzt = datetime.now() neue_state = {} for alarm in aktive_alarme: schluessel = alarm_schluessel(alarm) alter_eintrag = state.get(schluessel, {}) startzeit_text = alter_eintrag.get("startzeit") or jetzt.strftime("%Y-%m-%d %H:%M:%S") try: startzeit = datetime.strptime(startzeit_text, "%Y-%m-%d %H:%M:%S") except ValueError: startzeit = jetzt startzeit_text = jetzt.strftime("%Y-%m-%d %H:%M:%S") dauer = int((jetzt - startzeit).total_seconds()) eskalation, hinweise = eskalation_fuer_alarm(alarm, dauer) alarm["alarm_startzeit"] = startzeit_text alarm["alarmdauer_sekunden"] = dauer alarm["alarmdauer_minuten"] = round(dauer / 60, 1) alarm["alarmdauer_text"] = formatiere_alarmdauer(dauer) alarm["eskalationsstatus"] = eskalation alarm["eskalationshinweise"] = hinweise neue_state[schluessel] = { "startzeit": startzeit_text, "status": alarm.get("status"), "meldung": alarm.get("meldung"), "typ": alarm.get("typ", "motor"), } speichere_active_alarm_state(neue_state) return aktive_alarme def laengster_alarm(aktive_alarme): """Gibt den laengsten aktiven Alarm zurueck.""" if not aktive_alarme: return None return max(aktive_alarme, key=lambda alarm: alarm.get("alarmdauer_sekunden", 0)) def ergaenze_produktionsverlust_timer(produktion, aktive_alarme): """Verknuepft rote Alarme mit einem Demo-Produktionsverlust-Timer.""" rote_alarme = [alarm for alarm in aktive_alarme if alarm.get("status") == "Rot"] alarm = laengster_alarm(rote_alarme) if not alarm: produktion.update({ "produktionsverlust_aktiv": False, "produktionsverlust_seit": "-", "produktionsverlust_dauer_text": "-", "bisheriger_teileverlust": 0, "bisheriger_warenwertverlust": 0, }) return produktion dauer_stunden = alarm.get("alarmdauer_sekunden", 0) / 3600 teileverlust = round(produktion.get("teileverlust_pro_stunde", 0) * dauer_stunden, 1) warenwertverlust = round(teileverlust * produktion.get("wert_pro_teil", 0), 2) produktion.update({ "produktionsverlust_aktiv": True, "produktionsverlust_alarm": alarm.get("motorname"), "produktionsverlust_seit": alarm.get("alarm_startzeit"), "produktionsverlust_dauer_text": alarm.get("alarmdauer_text"), "bisheriger_teileverlust": teileverlust, "bisheriger_warenwertverlust": warenwertverlust, }) return produktion def erstelle_sensor_alarme(sensoren): """Erstellt aktive Sensoralarme fuer Dashboard und Quittierung.""" alarme = [] for sensor in sensoren: if sensor["status"] == "OK": continue alarme.append({ "typ": "sensor", "id": sensor["id"], "motorname": sensor["sensorname"], "bereich": sensor["bereich"], "status": "Rot" if sensor["status"] == "Fehler" else "Gelb", "meldungstyp": sensor["meldungstyp"], "risikowert": 60 if sensor["status"] == "Fehler" else 35, "meldung": sensor["hinweis"], "alarm_quittiert": sensor["alarm_quittiert"], }) return alarme def berechne_kosten_summe(berichte): """Summiert wirtschaftliche Demo-Werte fuer den Kostenbereich.""" return { "moegliche_ausfallkosten": sum( bericht["wirtschaftliche_bewertung"]["moegliche_ausfallkosten"] for bericht in berichte ), "geplante_wartungskosten": sum( bericht["wirtschaftliche_bewertung"]["geplante_wartungskosten"] for bericht in berichte ), "moegliches_einsparpotenzial": sum( bericht["wirtschaftliche_bewertung"]["moegliches_einsparpotenzial"] for bericht in berichte ), "mehrkosten_pro_tag": round(sum( bericht["energie_mehrkosten"]["mehrkosten_pro_tag"] for bericht in berichte ), 2), "mehrkosten_pro_monat": round(sum( bericht["energie_mehrkosten"]["mehrkosten_pro_monat"] for bericht in berichte ), 2), } def berechne_produktionsdaten(berichte, simulation_state=None): """Simuliert Demo-Produktionskennzahlen aus dem Anlagenzustand.""" simulation_state = simulation_state or lade_simulation_state() betriebsart = simulation_state.get("betriebsart", "automatik") sollleistung = PRODUKTIONS_GRUNDWERTE["sollleistung_pro_stunde"] wert_pro_teil = PRODUKTIONS_GRUNDWERTE["wert_pro_teil"] status_liste = [bericht["gesamtstatus"] for bericht in berichte] hat_rot = "Rot" in status_liste hat_gelb = "Gelb" in status_liste kritischer_hochrelevanter_motor = any( bericht["gesamtstatus"] == "Rot" and bericht["produktionsrelevanz"].startswith("Hoch") for bericht in berichte ) if betriebsart in ["wartung", "gestoppt"]: leistungsfaktor = 0.0 ausschussquote = 0.0 status = "gestoppt" status_text = "Produktion gestoppt - Startfreigabe gesperrt" elif betriebsart == "handbetrieb": leistungsfaktor = random.uniform(0.05, 0.15) ausschussquote = random.uniform(0.5, 1.5) status = "Handbetrieb / reduziert" status_text = "Handbetrieb aktiv - automatische Schrittfolge pausiert" elif hat_rot: leistungsfaktor = random.uniform(0.30, 0.50 if kritischer_hochrelevanter_motor else 0.65) ausschussquote = random.uniform(3.0, 8.0) status = "kritisch reduziert" status_text = "Produktionsverlust wahrscheinlich - Instandhaltung erforderlich" elif hat_gelb: leistungsfaktor = random.uniform(0.75, 0.90) ausschussquote = random.uniform(1.5, 3.0) status = "reduziert / beobachten" status_text = "Produktion reduziert - Anlage beobachten" else: leistungsfaktor = random.uniform(0.95, 1.05) ausschussquote = random.uniform(0.5, 1.5) status = "im Sollbereich" status_text = "Produktion laeuft im Sollbereich" aktuelle_leistung = round(sollleistung * leistungsfaktor) leistungsverlust = max(round((1 - aktuelle_leistung / sollleistung) * 100), 0) teileverlust_pro_stunde = max(sollleistung - aktuelle_leistung, 0) jetzt = datetime.now() stunden_in_schicht = (jetzt.hour % PRODUKTIONS_GRUNDWERTE["schichtdauer_stunden"]) + jetzt.minute / 60 produzierte_teile = round(aktuelle_leistung * stunden_in_schicht) ausschuss = round(produzierte_teile * ausschussquote / 100) gutteile = max(produzierte_teile - ausschuss, 0) moeglicher_teileverlust = teileverlust_pro_stunde * 4 warenwertverlust = moeglicher_teileverlust * wert_pro_teil return { "hinweis": "Demo-Produktionskennzahlen, keine echte Produktionsanbindung", "linie": PRODUKTIONS_GRUNDWERTE["linie"], "sollleistung_pro_stunde": sollleistung, "aktuelle_leistung_pro_stunde": aktuelle_leistung, "produzierte_teile_heute": produzierte_teile, "gutteile": gutteile, "ausschuss": ausschuss, "ausschussquote_prozent": round(ausschussquote, 1), "leistungsverlust_prozent": leistungsverlust, "status": status, "status_text": status_text, "betriebsart": betriebsart, "teileverlust_pro_stunde": teileverlust_pro_stunde, "moeglicher_teileverlust": moeglicher_teileverlust, "wert_pro_teil": wert_pro_teil, "moeglicher_warenwertverlust": warenwertverlust, } def erstelle_betriebsempfehlung(bericht): """Gibt eine einfache Demo-Empfehlung fuer den Weiterbetrieb.""" if bericht["gesamtstatus"] == "Gruen": return "Weiterbetrieb normal moeglich (Demo-Empfehlung)" if bericht["gesamtstatus"] == "Gelb": return "Weiterbetrieb moeglich, Werte beobachten (Demo-Empfehlung)" if bericht["risikowert"] >= 90: return "Produktionsstopp oder sofortige Pruefung pruefen (Demo-Empfehlung)" return "Weiterbetrieb nur mit Freigabe der Instandhaltung (Demo-Empfehlung)" def erstelle_aufgabenliste(berichte, sensoren): """Erzeugt automatisch einfache Aufgaben fuer die Instandhaltung.""" aufgaben = [] nummer = 1 for bericht in sorted(berichte, key=lambda eintrag: eintrag["risikowert"], reverse=True): if bericht["gesamtstatus"] == "Gruen": continue aufgaben.append({ "id": f"A-{nummer:03d}", "objekt": bericht["motorname"], "aufgabe": bericht["empfohlene_massnahmen"][0], "prioritaet": bericht["instandhaltungsprioritaet"], "status": "Offen", "wartungsfenster": bericht["empfohlenes_wartungsfenster"], "gewerk": bericht["empfohlenes_gewerk"], "erstellt_am": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }) nummer += 1 for sensor in sensoren: if sensor["status"] == "OK": continue aufgaben.append({ "id": f"A-{nummer:03d}", "objekt": sensor["sensorname"], "aufgabe": sensor["empfohlene_pruefung"][0], "prioritaet": "Prioritaet 2 - Zeitnah pruefen" if sensor["status"] == "Fehler" else "Prioritaet 3 - Einplanen", "status": "Offen", "wartungsfenster": "Naechstes verfuegbares Wartungsfenster", "gewerk": "Elektrik", "erstellt_am": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }) nummer += 1 return aufgaben def erstelle_schichtuebergabe_text(berichte, aktive_alarme, sensoren, aufgaben): """Erstellt ein Schichtuebergabe-Protokoll als Text.""" ranking = sorted(berichte, key=lambda eintrag: eintrag["risikowert"], reverse=True) kritischster = ranking[0] if ranking else None offene_alarme = [alarm for alarm in aktive_alarme if not alarm.get("alarm_quittiert")] quittierte_alarme = [alarm for alarm in aktive_alarme if alarm.get("alarm_quittiert")] reparaturen = lese_letzte_reparaturen(5) zeilen = [ "Schichtuebergabe - Predictive Maintenance Light", f"Datum: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", "", f"Gesamtstatus: {kritischster['gesamtstatus'] if kritischster else 'Gruen'}", f"Kritischster Motor: {kritischster['motorname'] if kritischster else '-'}", f"Risiko: {kritischster['risikowert'] if kritischster else 0} / 100", f"Offene Alarme: {len(offene_alarme)}", f"Quittierte, aktive Alarme: {len(quittierte_alarme)}", "", "Naechste Handlung:", ] if kritischster: zeilen.append(f"- {kritischster['motorname']} pruefen: {kritischster['wartungsempfehlung']}") zeilen.append(f"- Wartungsfenster: {kritischster['empfohlenes_wartungsfenster']}") zeilen.append(f"- Gewerk: {kritischster['empfohlenes_gewerk']}") zeilen.extend(["", "Aktive Alarme:"]) if aktive_alarme: for alarm in aktive_alarme: status = "quittiert" if alarm.get("alarm_quittiert") else "offen" codes = ", ".join(code["fehlercode"] for code in alarm.get("passende_fehlercodes", [])) or "ohne Fehlercode" dauer = alarm.get("alarmdauer_text", "-") eskalation = alarm.get("eskalationsstatus", "-") zeilen.append(f"- {alarm['motorname']}: {alarm['status']} - {alarm['meldung']} ({status}, {dauer}, {eskalation}, {codes})") else: zeilen.append("- Keine aktiven Alarme") zeilen.extend(["", "Dokumentierte Reparaturen:"]) if reparaturen: for eintrag in reparaturen: zeilen.append( f"- {eintrag.get('Motorname', '-')}: {eintrag.get('Durchgefuehrte_Massnahme', '-')} - " f"{eintrag.get('Ergebnis', '-')} - {eintrag.get('Mitarbeiter', '-')}" ) else: zeilen.append("- Keine Reparaturen dokumentiert") zeilen.extend(["", "Wichtige Hinweise:"]) for bericht in berichte: for warnung in bericht.get("nicht_nochmal_tauschen_warnungen", []): zeilen.append(f"- {bericht['motorname']}: {warnung['hinweis']}") for fehler in bericht.get("elektronische_fehler", []): zeilen.append(f"- {bericht['motorname']}: {fehler['diagnose']}") for warnung in bericht.get("plausibilitaets_warnungen", []): zeilen.append(f"- {bericht['motorname']}: {warnung}") zeilen.extend(["", "Offene Aufgaben:"]) if aufgaben: for aufgabe in aufgaben[:8]: zeilen.append(f"- {aufgabe['id']} {aufgabe['objekt']}: {aufgabe['aufgabe']} ({aufgabe['prioritaet']})") else: zeilen.append("- Keine offenen Aufgaben") text = "\n".join(zeilen) with open(SCHICHTUEBERGABE_PFAD, mode="w", encoding="utf-8") as datei: datei.write(text) return text def erstelle_tagesbericht_text(berichte, aktive_alarme, aufgaben): """Erzeugt eine einfache Tagesbericht-Datei.""" ranking = sorted(berichte, key=lambda eintrag: eintrag["risikowert"], reverse=True) kritischster = ranking[0] if ranking else None zaehler = erstelle_status_zaehler(berichte) kosten = berechne_kosten_summe(berichte) fehlercode_liste = [] for alarm in aktive_alarme: fehlercode_liste.extend(code["fehlercode"] for code in alarm.get("passende_fehlercodes", [])) fehlercode_text = ", ".join(fehlercode_liste) if fehlercode_liste else "Keine aktiven Fehlercodes" laengster = laengster_alarm(aktive_alarme) text = "\n".join([ f"Tagesbericht Predictive Maintenance Light - {datetime.now().strftime('%Y-%m-%d')}", f"Erstellt: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", "", f"Kritischster Motor: {kritischster['motorname'] if kritischster else '-'}", f"Höchstes Risiko: {kritischster['risikowert'] if kritischster else 0} / 100", f"Status: Gruen {zaehler['gruen']} | Gelb {zaehler['gelb']} | Rot {zaehler['rot']}", f"Aktive Alarme: {len(aktive_alarme)}", f"Laengster aktiver Alarm: {laengster['motorname']} - {laengster['alarmdauer_text']}" if laengster else "Laengster aktiver Alarm: keiner", f"Aktive Fehlercodes: {fehlercode_text}", f"Offene Aufgaben: {len(aufgaben)}", f"Einsparpotenzial gesamt: {kosten['moegliches_einsparpotenzial']} EUR", f"Energie-Mehrkosten pro Monat: {kosten['mehrkosten_pro_monat']} EUR", "", "Wichtigste Empfehlung:", kritischster["wartungsempfehlung"] if kritischster else "Keine Auffaelligkeit", ]) with open(TAGESBERICHT_PFAD, mode="w", encoding="utf-8") as datei: datei.write(text) return text def generate_produktionslinie(berichte, sensoren, schrittkette, simulation_state, energieversorgung): """Erzeugt die Produktionslinie-Datenstruktur (Linie 1 – Metallbearbeitung).""" m1 = next((m for m in berichte if m.get("motorname") == "Motor Foerderband 1"), berichte[0] if berichte else {}) m2 = next((m for m in berichte if m.get("motorname") == "Motor Pumpe 2"), berichte[1] if len(berichte) > 1 else {}) m3 = next((m for m in berichte if m.get("motorname") == "Motor Luefter 3"), berichte[2] if len(berichte) > 2 else {}) s1 = next((s for s in sensoren if s.get("id") == "FB1-S1"), None) s2 = next((s for s in sensoren if s.get("id") == "FB1-S2"), None) s3 = next((s for s in sensoren if s.get("id") == "FB1-S3"), None) modus = simulation_state.get("anlagenmodus", "standard") materialfluss = schrittkette.get("materialfluss", "aktiv") schritt = schrittkette.get("aktueller_schritt", 0) def _sc(gs): if gs == "Rot": return "Fehler" if gs == "Gelb": return "Warnung" return "OK" def _stub_sensorik(strom=8.0, temp=55.0, vib=1.5, drz=1450, bsh=0, last=72): return { "stromaufnahme": {"wert": strom, "einheit": "A", "status": "Gruen", "abw": 0.0}, "temperatur": {"wert": temp, "einheit": "°C", "status": "Gruen", "abw": 0.0}, "vibration": {"wert": vib, "einheit": "mm/s", "status": "Gruen", "abw": 0.0}, "drehzahl": {"wert": drz, "einheit": "U/min", "status": "Gruen", "abw": 0.0}, "betriebsstunden": bsh, "last_prozent": last, "fu_status": "OK", "drehzahl_rueckmeldung": "aktiv", "motorschutz_ok": True, } def _bericht_sensorik(bericht): if not bericht: return _stub_sensorik() messwerte = {e["messwert"]: e for e in bericht.get("messwerte", [])} def _mv(key, fallback=0.0): e = messwerte.get(key, {}) return { "wert": round(float(e.get("aktueller_wert", fallback)), 1), "status": e.get("status", "Gruen"), "abw": round(float(e.get("abweichung_prozent", 0.0)), 1), } strom_mv = _mv("Stromaufnahme", 8.0) last_est = min(100, round(70 + max(0, strom_mv["abw"]), 0)) gs = bericht.get("gesamtstatus", "Gruen") return { "stromaufnahme": {**_mv("Stromaufnahme", 8.0), "einheit": "A"}, "temperatur": {**_mv("Temperatur", 55.0), "einheit": "°C"}, "vibration": {**_mv("Vibration", 1.5), "einheit": "mm/s"}, "drehzahl": {**_mv("Drehzahl", 1450.0), "einheit": "U/min"}, "betriebsstunden": bericht.get("betriebsstunden", 0), "last_prozent": last_est, "fu_status": "Störung" if gs == "Rot" else "OK", "drehzahl_rueckmeldung": "fehlt" if gs == "Rot" else "aktiv", "motorschutz_ok": gs != "Rot", } band1_motorgs = m1.get("gesamtstatus", "Gruen") if m1 else "Gruen" band1_stau = modus == "materialstau" band1_status = "Fehler" if band1_motorgs == "Rot" else ("Warnung" if band1_stau or band1_motorgs == "Gelb" else "OK") linie_status = band1_status if band1_status != "OK" else "OK" material_auf_band1 = 4 <= schritt <= 8 # ── Simulationseffekte auf neue Stationskomponenten ────────────────────── def _sim_adj(basis_risiko, motor_id): """Passt Risikowert und Status je nach Simulationsmodus an.""" r = basis_risiko if modus == "normal": r = max(0, basis_risiko // 3) elif modus in ["auffaellig", "wiederholfehler"]: r = min(55, basis_risiko + 18) elif modus == "kritisch": if motor_id in ("SAEGE_M1", "CNC_SP1"): r = 72 elif motor_id in ("CNC_SP2", "SAEGE_M2", "KUKA_R1"): r = 58 else: r = min(45, basis_risiko + 25) elif modus in ("versorgungsfehler", "400v_fehler"): if motor_id.startswith("CNC_"): r = min(65, basis_risiko + 35) elif motor_id.startswith("SAEGE_"): r = min(50, basis_risiko + 22) elif modus == "fu_stoerung": if motor_id in ("BAND_2_M1", "BAND_3_M1"): r = min(48, basis_risiko + 28) elif modus == "materialstau": if motor_id in ("BAND_2_M1", "BAND_1_M1"): r = min(52, basis_risiko + 32) elif modus == "cnc_stoerung": if motor_id == "CNC_SP2": r = 78 elif motor_id.startswith("CNC_"): r = min(55, basis_risiko + 28) elif modus == "saege_stoerung": if motor_id == "SAEGE_M1": r = 72 elif motor_id.startswith("SAEGE_"): r = min(50, basis_risiko + 22) elif modus == "reinigung_stoerung": if motor_id == "REIN_M1": r = 65 elif motor_id.startswith("REIN_"): r = min(48, basis_risiko + 20) elif modus == "reinigung_druckverlust": if motor_id == "REIN_M1": r = 55 elif modus == "roboter_stoerung": if motor_id in ("KUKA_R1", "KUKA_R2"): r = 68 elif modus == "spindel_lager_warnung": if motor_id == "CNC_SP2": r = 62 elif motor_id in ("CNC_SP1", "CNC_SP3"): r = min(45, basis_risiko + 20) elif modus == "achse_schwergaengig": if motor_id.startswith("CNC_"): r = min(50, basis_risiko + 25) elif modus == "sicherheit_not_aus": r = 0 elif modus == "medien_luftdruck_niedrig": r = min(35, basis_risiko + 12) s = "Fehler" if r >= 75 else "Warnung" if r >= 40 else "OK" return r, s _saege_m1_r, _saege_m1_s = _sim_adj(12, "SAEGE_M1") _saege_m2_r, _saege_m2_s = _sim_adj(8, "SAEGE_M2") _saege_m3_r, _saege_m3_s = _sim_adj(5, "SAEGE_M3") _cnc_sp1_r, _cnc_sp1_s = _sim_adj(18, "CNC_SP1") _cnc_sp2_r, _cnc_sp2_s = _sim_adj(22, "CNC_SP2") _cnc_sp3_r, _cnc_sp3_s = _sim_adj(14, "CNC_SP3") _cnc_sp4_r, _cnc_sp4_s = _sim_adj(9, "CNC_SP4") _rein_m1_r, _rein_m1_s = _sim_adj(15, "REIN_M1") _rein_m2_r, _rein_m2_s = _sim_adj(10, "REIN_M2") _rein_m3_r, _rein_m3_s = _sim_adj(7, "REIN_M3") _kuka_r1_r, _kuka_r1_s = _sim_adj(8, "KUKA_R1") _kuka_r2_r, _kuka_r2_s = _sim_adj(6, "KUKA_R2") # Saegestatus aus Simulation ableiten _saege_status = "Fehler" if _saege_m1_s == "Fehler" else "Warnung" if _saege_m1_s == "Warnung" else "OK" _cnc_status = "Fehler" if _cnc_sp1_s == "Fehler" or _cnc_sp2_s == "Fehler" else "Warnung" if _cnc_sp1_s == "Warnung" else "OK" _rein_status = "Fehler" if _rein_m1_s == "Fehler" else "Warnung" if _rein_m1_s == "Warnung" else "OK" _versand_status = "Fehler" if _kuka_r2_s == "Fehler" else "Warnung" if _kuka_r2_s == "Warnung" else "OK" # Update linie_status to include new stations if _cnc_status == "Fehler" or _saege_status == "Fehler" or _rein_status == "Fehler" or _versand_status == "Fehler": linie_status = "Fehler" elif linie_status == "OK" and any(s == "Warnung" for s in [_saege_status, _cnc_status, _rein_status, _versand_status]): linie_status = "Warnung" linie = { "id": "linie_1", "name": "Linie 1 — Metallbearbeitung", "status": linie_status, "betriebsart": simulation_state.get("betriebsart", "automatik"), "materialfluss": materialfluss, "aktuelle_station": "BAND_1" if material_auf_band1 else "materialeingang", "engpass": "CNC" if _cnc_status == "Fehler" else "BAND_1" if band1_stau else "SAEGE" if _saege_status == "Fehler" else "-", "stationen": [ {"id": "materialeingang", "name": "Materialeingang", "kurzname": "Eingang", "status": "OK", "aktuelle_aktivitaet": "Bereit für nächsten Metallblock", "typ": "eingang", "motoren": [], "sensoren": [], "roboter": None, "aktiver_alarm": None, "produktionsauswirkung": "Keine", "position_in_linie": 0}, {"id": "saegestation", "name": "Sägestation", "kurzname": "Sägen", "status": _saege_status, "aktuelle_aktivitaet": "Bereit" if _saege_status == "OK" else "Störung", "typ": "bearbeitungsstation", "motoren": [ {"id": "SAEGE_M1", "name": "Hauptsägeantrieb", "status": _saege_m1_s, "risikowert": _saege_m1_r, "sensorik": _stub_sensorik(strom=18.5, temp=62.0, vib=2.8, drz=2950, bsh=2340, last=68)}, {"id": "SAEGE_M2", "name": "Vorschub / Positionierung", "status": _saege_m2_s, "risikowert": _saege_m2_r, "sensorik": _stub_sensorik(strom=6.2, temp=48.0, vib=1.2, drz=1450, bsh=2340, last=55)}, {"id": "SAEGE_M3", "name": "Kühlung / Absaugung", "status": _saege_m3_s, "risikowert": _saege_m3_r, "sensorik": _stub_sensorik(strom=4.8, temp=44.0, vib=0.9, drz=2900, bsh=2340, last=48)}, ], "sensoren": ["SAEGE_S1", "SAEGE_S2", "SAEGE_S3", "SAEGE_S4"], "roboter": None, "aktiver_alarm": None, "produktionsauswirkung": "Keine" if _saege_status == "OK" else "Sägebetrieb unterbrochen — Materialzufuhr gestoppt", "position_in_linie": 2}, {"id": "cnc_station", "name": "CNC-Frässtation", "kurzname": "CNC-Fräsen", "status": _cnc_status, "aktuelle_aktivitaet": "Bereit" if _cnc_status == "OK" else "Störung", "typ": "cnc", "motoren": [ {"id": "CNC_SP1", "name": "Spindel 1", "status": _cnc_sp1_s, "risikowert": _cnc_sp1_r, "sensorik": _stub_sensorik(strom=22.0, temp=71.0, vib=3.5, drz=8000, bsh=1820, last=74)}, {"id": "CNC_SP2", "name": "Spindel 2", "status": _cnc_sp2_s, "risikowert": _cnc_sp2_r, "sensorik": _stub_sensorik(strom=24.5, temp=74.0, vib=3.9, drz=8000, bsh=1820, last=79)}, {"id": "CNC_SP3", "name": "Spindel 3", "status": _cnc_sp3_s, "risikowert": _cnc_sp3_r, "sensorik": _stub_sensorik(strom=19.8, temp=68.0, vib=3.1, drz=8000, bsh=1820, last=71)}, {"id": "CNC_SP4", "name": "Spindel 4", "status": _cnc_sp4_s, "risikowert": _cnc_sp4_r, "sensorik": _stub_sensorik(strom=16.2, temp=62.0, vib=2.4, drz=8000, bsh=1820, last=65)}, ], "sensoren": ["CNC_S1", "CNC_S2", "CNC_S3", "CNC_S4"], "roboter": None, "aktiver_alarm": None, "produktionsauswirkung": "Keine" if _cnc_status == "OK" else "CNC-Bearbeitung unterbrochen — Ausschussrisiko hoch", "position_in_linie": 4}, {"id": "reinigungsstation", "name": "Reinigungsstation", "kurzname": "Reinigung", "status": _rein_status, "aktuelle_aktivitaet": "Bereit" if _rein_status == "OK" else "Störung", "typ": "reinigung", "motoren": [ {"id": "REIN_M1", "name": "Pumpe", "status": _rein_m1_s, "risikowert": _rein_m1_r, "sensorik": _stub_sensorik(strom=9.5, temp=57.0, vib=2.1, drz=2900, bsh=3100, last=62)}, {"id": "REIN_M2", "name": "Fördertechnik", "status": _rein_m2_s, "risikowert": _rein_m2_r, "sensorik": _stub_sensorik(strom=7.8, temp=52.0, vib=1.6, drz=1450, bsh=3100, last=58)}, {"id": "REIN_M3", "name": "Trocknung / Lüfter", "status": _rein_m3_s, "risikowert": _rein_m3_r, "sensorik": _stub_sensorik(strom=5.5, temp=46.0, vib=1.1, drz=2900, bsh=3100, last=51)}, ], "sensoren": ["REIN_S1", "REIN_S2", "REIN_S3"], "roboter": {"id": "KUKA_R1", "name": "KUKA R1", "aufgabe": "Handling nach Reinigung", "status": _kuka_r1_s, "programmstatus": "Bereit" if _kuka_r1_s == "OK" else "Alarm", "greiferstatus": "Offen", "sensorik": _stub_sensorik(strom=14.2, temp=58.0, vib=2.2, drz=0, bsh=1560, last=60), "risikowert": _kuka_r1_r}, "aktiver_alarm": None, "produktionsauswirkung": "Keine" if _rein_status == "OK" else "Reinigungsprozess unterbrochen — Teile bleiben nass", "position_in_linie": 6}, {"id": "schwarzkammer", "name": "Schwarzkammer / Prüfstation", "kurzname": "Prüfung", "status": "OK", "aktuelle_aktivitaet": "Bereit", "typ": "pruefstation", "motoren": [], "sensoren": ["PRUEF_S1", "PRUEF_S2", "PRUEF_S3", "PRUEF_S4", "PRUEF_S5"], "roboter": None, "aktiver_alarm": None, "produktionsauswirkung": "Keine", "position_in_linie": 8, "letztes_ergebnis": "Gutteil", "gutteil_quote": 98.5}, {"id": "versand", "name": "Versand / Verpackung", "kurzname": "Versand", "status": _versand_status, "aktuelle_aktivitaet": "Bereit" if _versand_status == "OK" else "Störung", "typ": "versand", "motoren": [], "sensoren": ["VERSAND_S1", "VERSAND_S2"], "roboter": {"id": "KUKA_R2", "name": "KUKA R2", "aufgabe": "Verpackung", "status": _kuka_r2_s, "programmstatus": "Bereit" if _kuka_r2_s == "OK" else "Alarm", "greiferstatus": "Offen", "sensorik": _stub_sensorik(strom=12.8, temp=55.0, vib=1.9, drz=0, bsh=980, last=55), "risikowert": _kuka_r2_r}, "aktiver_alarm": None, "produktionsauswirkung": "Keine" if _versand_status == "OK" else "Versand blockiert — fertige Teile stauen vor Versand", "position_in_linie": 10}, ], "foerdermodule": [ {"id": "BAND_1", "name": "Förderband 1", "langname": "Materialeingang → Sägestation", "von_station": "materialeingang", "nach_station": "saegestation", "status": band1_status, "materialfluss": materialfluss, "material_vorhanden": material_auf_band1, "material_position": "zwischen S1 und S2" if material_auf_band1 else "-", "stau": band1_stau, "motoren": [{"id": "BAND_1_M1", "name": "M1 — Förderbandantrieb", "status": _sc(band1_motorgs), "risikowert": m1.get("risikowert", 0) if m1 else 0, "motorname_original": m1.get("motorname", "") if m1 else "", "sensorik": _bericht_sensorik(m1)}], "nebenaggregate": [ {"id": "BAND_1_M2", "name": "M2 — Pumpe / Kühlung", "status": _sc(m2.get("gesamtstatus", "Gruen") if m2 else "Gruen"), "risikowert": m2.get("risikowert", 0) if m2 else 0, "sensorik": _bericht_sensorik(m2)}, {"id": "BAND_1_M3", "name": "M3 — Lüfter / Umgebung", "status": _sc(m3.get("gesamtstatus", "Gruen") if m3 else "Gruen"), "risikowert": m3.get("risikowert", 0) if m3 else 0, "sensorik": _bericht_sensorik(m3)}, ], "sensoren": [ {"id": "BAND_1_S1", "name": "S1 — Lichtschranke Eingang", "status": s1.get("status", "OK") if s1 else "OK", "sensor_id_original": "FB1-S1"}, {"id": "BAND_1_S2", "name": "S2 — Lichtschranke Ausgang", "status": s2.get("status", "OK") if s2 else "OK", "sensor_id_original": "FB1-S2"}, {"id": "BAND_1_S3", "name": "S3 — Bandlaufüberwachung", "status": s3.get("status", "OK") if s3 else "OK", "sensor_id_original": "FB1-S3"}, ], "fehlercodes": [fc.get("fehlercode", "") for fc in (m1.get("passende_fehlercodes", []) if m1 else [])], "empfohlene_massnahme": m1.get("empfohlene_massnahme", "Keine Maßnahme erforderlich") if m1 else "Keine Maßnahme erforderlich", "uses_existing_detail": True}, *[{"id": f"BAND_{i}", "name": f"Förderband {i}", "langname": name, "von_station": von, "nach_station": nach, "status": _sim_adj(5, f"BAND_{i}_M1")[1], "materialfluss": "aktiv" if modus not in ("fu_stoerung", "materialstau") else "steht", "material_vorhanden": False, "material_position": "-", "stau": False, "motoren": [{"id": f"BAND_{i}_M1", "name": "M1 — Förderbandantrieb", "status": _sim_adj(5, f"BAND_{i}_M1")[1], "risikowert": _sim_adj(5, f"BAND_{i}_M1")[0], "sensorik": _stub_sensorik(strom=8.0, temp=52.0, vib=1.4, drz=1450, bsh=0, last=68)}], "nebenaggregate": [], "sensoren": [ {"id": f"BAND_{i}_S1", "name": "S1 — Lichtschranke Eingang", "status": "OK"}, {"id": f"BAND_{i}_S2", "name": "S2 — Lichtschranke Ausgang", "status": "OK"}, {"id": f"BAND_{i}_S3", "name": "S3 — Bandlaufüberwachung", "status": "OK"}, ], "fehlercodes": [], "empfohlene_massnahme": "Keine Maßnahme erforderlich", "uses_existing_detail": False} for i, (name, von, nach) in enumerate([ ("Sägestation → CNC-Frässtation", "saegestation", "cnc_station"), ("CNC-Frässtation → Reinigungsstation", "cnc_station", "reinigungsstation"), ("Reinigungsstation → Schwarzkammer", "reinigungsstation", "schwarzkammer"), ("Schwarzkammer → Versand", "schwarzkammer", "versand"), ], start=2)], ], "linie_alarme": [], "linie_produktion": { "teile_heute": 0, "gutteile": 0, "ausschuss": 0, "oee_light": "-", "taktzeit_soll": "45s", "engpass_station": None, }, } stundenfaktor = max(1, datetime.now().hour or 1) basis_produktion = [max(20, int(stundenfaktor * 11)), max(18, int(stundenfaktor * 10.5)), max(16, int(stundenfaktor * 10))] for index, station in enumerate(linie["stationen"]): status = station.get("status", "OK") abschlag = 0.65 if status == "Fehler" else 0.82 if status == "Warnung" else 1.0 teile = round(basis_produktion[index % len(basis_produktion)] * abschlag) station.setdefault("produktion", {}) station["produktion"].setdefault("teile_heute", teile) stationen = linie["stationen"] teile_heute = sum(s.get("produktion", {}).get("teile_heute", 0) for s in stationen) warnungen = sum(1 for s in stationen if s.get("status") == "Warnung") fehler_oder_warnungen = sum(1 for s in stationen if s.get("status") != "OK") linie["linie_produktion"] = { "teile_heute": teile_heute, "teile_pro_stunde": round(teile_heute / max(1, datetime.now().hour or 1), 1), "oee_light": round(90 - fehler_oder_warnungen * 8, 1), "taktzeit_soll": 45, "taktzeit_ist": 45 + sum(2 for s in stationen if s.get("status") == "Warnung"), "gutteile": max(0, teile_heute - 2), "ausschuss": 2, "engpass_station": "BAND_1" if band1_stau else None, } return linie def _extrahiere_alle_motoren(linie): """Gibt alle Motoren aus der Produktionslinie als flache Liste zurueck.""" result = [] for station in linie.get("stationen", []): for motor in station.get("motoren", []): result.append({ **motor, "station_id": station["id"], "station_name": station["name"], "bereich": station.get("typ", "unbekannt"), "objekt_typ": "motor", }) roboter = station.get("roboter") if roboter: result.append({ **roboter, "station_id": station["id"], "station_name": station["name"], "bereich": station.get("typ", "unbekannt"), "objekt_typ": "roboter", }) for band in linie.get("foerdermodule", []): for motor in band.get("motoren", []): result.append({ **motor, "station_id": band["id"], "station_name": band["name"], "bereich": "foerdertechnik", "objekt_typ": "motor", }) for motor in band.get("nebenaggregate", []): result.append({ **motor, "station_id": band["id"], "station_name": band["name"], "bereich": "foerdertechnik", "objekt_typ": "motor", }) return result def _ergaenze_alarm_stationen(aktive_alarme, alle_motoren): """Ergaenzt Alarmen Stationsreferenzen, ohne bestehende Alarmfelder zu veraendern.""" lookup = {} fb1_station = next((motor for motor in alle_motoren if motor.get("station_id") == "BAND_1"), None) for motor in alle_motoren: for key in [ motor.get("id"), motor.get("name"), motor.get("motorname"), motor.get("motorname_original"), ]: if key: lookup[key] = motor if motor.get("id") == "BAND_1_M2": lookup["Motor Pumpe 2"] = motor if motor.get("id") == "BAND_1_M3": lookup["Motor Luefter 3"] = motor if motor.get("id") == "BAND_1_M1": lookup["Motor Foerderband 1"] = motor for alarm in aktive_alarme: motor = lookup.get(alarm.get("id")) or lookup.get(alarm.get("motorname")) if motor: alarm.setdefault("station_id", motor.get("station_id")) alarm.setdefault("station_name", motor.get("station_name")) elif fb1_station: alarm_text = " ".join(str(alarm.get(key, "")) for key in ["id", "motorname", "fehlercode", "meldung"]) if any(token in alarm_text for token in ["S1", "S2", "S3", "I0.", "Q0.", "IW", "SPS", "400V", "24V"]): alarm.setdefault("station_id", fb1_station.get("station_id")) alarm.setdefault("station_name", fb1_station.get("station_name")) return aktive_alarme def speichere_dashboard_daten(berichte, sensoren=None, sps_io=None, schrittkette=None): """Speichert aktuelle Daten fuer das Dashboard als JSON.""" sensoren = sensoren or [] sps_io = sps_io or { "hinweis": "SPS I/O-Daten simuliert - keine echte SPS-Verbindung", "eingänge_digital": [], "eingänge_analog": [], "ausgänge_digital": [], "ausgänge_analog": [], "diagnosen": [], } schrittkette = schrittkette or { "hinweis": "Schrittkette simuliert", "name": "Foerderband 1 Ablauf", "aktueller_schritt": 0, "aktueller_schritt_name": "Grundstellung", "status": "Wartet", "stoerung": False, "ablaufdiagnose": "Noch keine Daten", "schritte": [], } simulation_state = lade_simulation_state() energieversorgung = erstelle_energieversorgung(simulation_state, berichte, sensoren) datenqualitaet = berechne_datenqualitaet(berichte, sensoren, sps_io, energieversorgung) materialfluss_detail = ermittle_materialfluss_detail(schrittkette, simulation_state) diagnose_motorstart = diagnostiziere_motorstart(sps_io, schrittkette) _linie_data = generate_produktionslinie(berichte, sensoren, schrittkette, simulation_state, energieversorgung) alle_motoren = _extrahiere_alle_motoren(_linie_data) aktive_alarme = erstelle_alarme(berichte) + erstelle_sensor_alarme(sensoren) + erstelle_sps_alarme(sps_io) aktive_alarme = _ergaenze_alarm_stationen(aktive_alarme, alle_motoren) ergaenze_passende_fehlercodes(berichte, sensoren, sps_io, schrittkette, aktive_alarme) aktive_alarme = aktualisiere_alarmdauer(aktive_alarme) aufgaben = erstelle_aufgabenliste(berichte, sensoren) praesentation_schritt = simulation_state.get("praesentation_schritt", 0) praesentation_info = PRAESENTATIONS_SCHRITTE[praesentation_schritt] produktion = berechne_produktionsdaten(berichte, simulation_state) produktion = ergaenze_produktionsverlust_timer(produktion, aktive_alarme) schichtuebergabe = erstelle_schichtuebergabe_text(berichte, aktive_alarme, sensoren, aufgaben) tagesbericht = erstelle_tagesbericht_text(berichte, aktive_alarme, aufgaben) aktive_fehlercodes = [] for alarm in aktive_alarme: for fehlercode in alarm.get("passende_fehlercodes", []): aktive_fehlercodes.append({ **fehlercode, "objekt": alarm.get("motorname"), "status": alarm.get("status"), "alarmdauer_text": alarm.get("alarmdauer_text"), "eskalationsstatus": alarm.get("eskalationsstatus"), "alarm_quittiert": alarm.get("alarm_quittiert"), }) aktive_fehlercode_ids = [code["fehlercode"] for code in aktive_fehlercodes if code.get("fehlercode")] produktion = erweitere_produktionsdaten( produktion, berichte, schrittkette, materialfluss_detail, energieversorgung, aktive_fehlercode_ids, ) signalwege = erstelle_signalwege( datenqualitaet, energieversorgung, sensoren, sps_io, aktive_fehlercode_ids, diagnose_motorstart, ) sps_daten = erzeuge_sps_daten(simulation_state) sicherheit_daten = erzeuge_sicherheit_daten(simulation_state) medien_daten = erzeuge_medien_daten(simulation_state) achsen_daten = erzeuge_achsen_daten(simulation_state) reinigung_detail = erzeuge_reinigung_detail(simulation_state) gesamtstatus_z = { "gesamt": len(alle_motoren), "ok": sum(1 for m in alle_motoren if m.get("status", "OK") == "OK"), "warnung": sum(1 for m in alle_motoren if m.get("status") == "Warnung"), "fehler": sum(1 for m in alle_motoren if m.get("status") == "Fehler"), } daten = { "titel": "Predictive Maintenance Light", "letzte_aktualisierung": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "settings": SETTINGS, "simulation_state": simulation_state, "praesentation": { "aktiv": simulation_state.get("praesentation_aktiv", False), "schritt": praesentation_schritt, "anzahl_schritte": len(PRAESENTATIONS_SCHRITTE), "titel": praesentation_info["titel"], "betroffen": praesentation_info["betroffen"], "auswirkung": praesentation_info["auswirkung"], "schritte": [ {"nummer": index + 1, "titel": eintrag["titel"]} for index, eintrag in enumerate(PRAESENTATIONS_SCHRITTE) ], }, "status_zaehler": erstelle_status_zaehler(berichte), "motoren": berichte, "alle_motoren": alle_motoren, "sps": sps_daten, "sicherheit": sicherheit_daten, "medien": medien_daten, "achsen": achsen_daten, "reinigung_detail": reinigung_detail, "gesamtstatus_zaehler": gesamtstatus_z, "produktionslinie": _linie_data, "aktive_alarme": aktive_alarme, "laengster_alarm": laengster_alarm(aktive_alarme), "fehlercode_katalog": FEHLERCODE_KATALOG, "aktive_fehlercodes": aktive_fehlercodes, "signalwege": signalwege, "kosten_summe": berechne_kosten_summe(berichte), "produktion": produktion, "ereignisprotokoll": lese_letzte_ereignisse(), "reparaturhistorie": lese_letzte_reparaturen(), "sensoren": sensoren, "energieversorgung": energieversorgung, "datenqualitaet": datenqualitaet, "materialfluss_detail": materialfluss_detail, "diagnose_motorstart": diagnose_motorstart, "ersatzteilbestand": ERSATZTEILBESTAND, "sps_io": sps_io, "schrittkette": schrittkette, "aufgaben": aufgaben, "schichtuebergabe": schichtuebergabe, "tagesbericht": tagesbericht, "schichtuebergabe_datei": SCHICHTUEBERGABE_DATEI, "tagesbericht_datei": TAGESBERICHT_DATEI, } schreibe_json_datei(DATEN_PFAD, daten) def erstelle_sicheren_dateinamen(text): """Erstellt einen einfachen Dateinamen aus einem Motor- oder Berichtsnamen.""" text = text.lower().replace(" ", "_") sichere_zeichen = [] for zeichen in text: if zeichen.isalnum() or zeichen == "_": sichere_zeichen.append(zeichen) return "".join(sichere_zeichen) def exportiere_wartungsbericht(bericht): """Erzeugt automatisch einen einfachen Wartungsbericht als Textdatei.""" if bericht["gesamtstatus"] != "Rot" and bericht["risikowert"] <= 60: return dateiname = f"wartungsbericht_{erstelle_sicheren_dateinamen(bericht['motorname'])}.txt" pfad = os.path.join(SCRIPT_ORDNER, dateiname) wirtschaft = bericht["wirtschaftliche_bewertung"] akte = bericht["maschinenakte"] with open(pfad, mode="w", encoding="utf-8") as datei: datei.write("Predictive Maintenance Light - Wartungsbericht\n") datei.write("=" * 56 + "\n") datei.write(f"Motorname: {bericht['motorname']}\n") datei.write(f"Bereich: {bericht['bereich']}\n") datei.write(f"Zeitstempel: {bericht['zeitstempel']}\n") datei.write(f"Gesamtstatus: {bericht['gesamtstatus']}\n") datei.write(f"Meldungstyp: {bericht['meldungstyp']}\n") datei.write(f"Prioritaet: {bericht['instandhaltungsprioritaet']}\n") datei.write(f"Risikowert: {bericht['risikowert']} / 100\n") datei.write(f"Ausfallprognose: {bericht['ausfallprognose_text']}\n") datei.write(f"Wartungsfenster: {bericht['empfohlenes_wartungsfenster']}\n\n") datei.write("Empfohlene Massnahmen:\n") for massnahme in bericht["empfohlene_massnahmen"]: datei.write(f"- {massnahme}\n") datei.write("\nErsatzteile / Pruefbereiche:\n") for empfehlung in bericht["ersatzteile_pruefbereiche"]: datei.write(f"- {empfehlung}\n") datei.write("\nWirtschaftliche Bewertung (Demo-Schaetzung):\n") datei.write(f"- Stillstandskosten pro Stunde: {wirtschaft['stillstandskosten_pro_stunde']} EUR\n") datei.write(f"- Ausfalldauer ohne Fruehwarnung: {wirtschaft['ausfalldauer_ohne_fruehwarnung']} h\n") datei.write(f"- Moegliche Ausfallkosten: {wirtschaft['moegliche_ausfallkosten']} EUR\n") datei.write(f"- Geplante Wartungskosten: {wirtschaft['geplante_wartungskosten']} EUR\n") datei.write(f"- Moegliches Einsparpotenzial: {wirtschaft['moegliches_einsparpotenzial']} EUR\n") datei.write("\nMaschinenakte:\n") datei.write(f"- Anlagenteil: {akte['anlagenteil']}\n") datei.write(f"- Leistung: {akte['leistung']}\n") datei.write(f"- Baujahr: {akte['baujahr']}\n") datei.write(f"- Letzte Wartung: {akte['letzte_wartung']}\n") datei.write(f"- Naechste geplante Wartung: {akte['naechste_geplante_wartung']}\n") datei.write(f"- Inventarnummer: {akte['inventarnummer']}\n") def reset_historie(): """Setzt die CSV-Historie auf die Kopfzeile zurueck.""" with open(HISTORIE_PFAD, mode="w", encoding="utf-8", newline="") as datei: writer = csv.writer(datei) writer.writerow(HISTORIE_FELDNAMEN) def reset_quittierungen(): """Leert gespeicherte Alarm-Quittierungen.""" speichere_quittierungen({}) def drucke_bericht(bericht): """Gibt einen kurzen Bericht in der Konsole aus.""" print("=" * 72) print(f"Motor / Anlage: {bericht['motorname']}") print(f"Zeitstempel: {bericht['zeitstempel']}") print(f"Gesamtstatus: {bericht['gesamtstatus']}") print(f"Meldungstyp: {bericht['meldungstyp']}") print(f"Prioritaet: {bericht['instandhaltungsprioritaet']}") print(f"Risikowert: {bericht['risikowert']} / 100") print(f"Langzeit: {bericht['langzeittrend']}") print("-" * 72) for ergebnis in bericht["messwerte"]: print( f"{ergebnis['messwert']:<15} " f"Normal: {ergebnis['normalwert']:>8} {ergebnis['einheit']:<5} " f"Aktuell: {ergebnis['aktueller_wert']:>8} {ergebnis['einheit']:<5} " f"Abw.: {ergebnis['abweichung_prozent']:>6.1f} % " f"Status: {ergebnis['status']}" ) print("Moegliche Ursache:") for ursache in bericht["moegliche_ursachen"]: print(f"- {ursache}") print() def erstelle_dashboard_html(): """Erstellt die HTML-Datei fuer das lokale Dashboard.""" if os.path.exists(DASHBOARD_PFAD): print("dashboard.html existiert bereits - bestehende Frontend-Dateien werden nicht ueberschrieben.") return print("dashboard.html fehlt. Der alte Fallback-Generator ist deaktiviert, damit keine M1/M2/M3-Demo die aktuelle Web-App ersetzt.") print("Bitte die aktuelle dashboard.html aus dem Projektstand wiederherstellen.") return html_inhalt = """ Predictive Maintenance Light

Predictive Maintenance Light

Lokales Mini-HMI Dashboard für Motoren und Antriebe

Letzte Aktualisierung -
""" with open(DASHBOARD_PFAD, mode="w", encoding="utf-8") as datei: datei.write(html_inhalt) def erstelle_style_css(): """Erstellt das dunkle Industrie-Design fuer das Dashboard.""" if os.path.exists(STYLE_PFAD): print("style.css existiert bereits - bestehende Frontend-Dateien werden nicht ueberschrieben.") return print("style.css fehlt. Der alte Fallback-Generator ist deaktiviert, damit kein veraltetes Design erzeugt wird.") print("Bitte die aktuelle style.css aus dem Projektstand wiederherstellen.") return css_inhalt = """:root { --hintergrund: #07111f; --karte: #101d2e; --karte-hell: #162840; --linie: #2b3f5b; --text: #ecf2f8; --text-muted: #9fb0c3; --gruen: #2ecc71; --gelb: #f1c40f; --rot: #e74c3c; } * { box-sizing: border-box; } body { margin: 0; min-height: 100vh; background: var(--hintergrund); color: var(--text); font-family: Arial, Helvetica, sans-serif; } .kopfbereich { display: flex; justify-content: space-between; align-items: flex-end; gap: 24px; padding: 24px 28px; border-bottom: 1px solid var(--linie); background: #0b1728; } h1, h2, h3, p { margin: 0; } h1 { font-size: 30px; } .kopfbereich p, .karten-kopf p, .aktualisierung span { color: var(--text-muted); } .aktualisierung { text-align: right; } .aktualisierung strong { display: block; margin-top: 6px; font-size: 18px; } .motor-grid { display: grid; grid-template-columns: repeat(3, minmax(320px, 1fr)); gap: 18px; padding: 22px; } .motor-karte { border: 1px solid var(--linie); border-top-width: 4px; border-radius: 8px; background: var(--karte); box-shadow: 0 12px 30px rgba(0, 0, 0, 0.22); overflow: hidden; } .karten-kopf { display: flex; justify-content: space-between; gap: 18px; align-items: center; padding: 18px; background: var(--karte-hell); } .karten-kopf h2 { font-size: 20px; margin-bottom: 8px; } .ampel { width: 58px; height: 58px; border-radius: 50%; flex: 0 0 auto; box-shadow: 0 0 28px currentColor; } .status-zeile { display: flex; justify-content: space-between; align-items: center; padding: 14px 18px; border-bottom: 1px solid var(--linie); } .status-zeile span { color: var(--text-muted); } table { width: 100%; border-collapse: collapse; font-size: 14px; } th, td { padding: 10px 12px; border-bottom: 1px solid rgba(159, 176, 195, 0.18); text-align: left; } th { color: var(--text-muted); font-weight: 600; } .status-chip { display: inline-block; min-width: 54px; padding: 4px 8px; border-radius: 4px; color: #07111f; font-weight: 700; text-align: center; } .balken-zeile td { padding-top: 0; } .balken-hintergrund { height: 9px; border-radius: 999px; background: #24364f; overflow: hidden; } .balken { height: 100%; min-width: 2px; } .info-bereich, .trend-bereich { padding: 16px 18px; border-top: 1px solid rgba(159, 176, 195, 0.18); } .info-bereich h3, .trend-bereich h3 { margin-bottom: 10px; color: var(--text-muted); font-size: 14px; text-transform: uppercase; } ul { margin: 0; padding-left: 18px; } li { margin-bottom: 7px; } .status-gruen { color: var(--gruen); background-color: var(--gruen); border-top-color: var(--gruen); } .status-gelb { color: var(--gelb); background-color: var(--gelb); border-top-color: var(--gelb); } .status-rot { color: var(--rot); background-color: var(--rot); border-top-color: var(--rot); } strong.status-gruen, strong.status-gelb, strong.status-rot { background: transparent; } .motor-karte.status-gruen, .motor-karte.status-gelb, .motor-karte.status-rot { background: var(--karte); color: var(--text); } @media (max-width: 1120px) { .motor-grid { grid-template-columns: 1fr; } } @media (max-width: 720px) { .kopfbereich { align-items: flex-start; flex-direction: column; } .aktualisierung { text-align: left; } .motor-grid { padding: 12px; } table { font-size: 12px; } th, td { padding: 8px 6px; } } """ with open(STYLE_PFAD, mode="w", encoding="utf-8") as datei: datei.write(css_inhalt) def erstelle_projektdateien(): """Erstellt fehlende Projektdateien. Bestehende Frontend-Dateien werden nicht ueberschrieben.""" # Bestehende Frontend-Dateien werden nicht ueberschrieben. if not os.path.exists(DASHBOARD_PFAD): erstelle_dashboard_html() if not os.path.exists(STYLE_PFAD): erstelle_style_css() if not os.path.exists(HISTORIE_PFAD): reset_historie() if not os.path.exists(ALARM_HISTORIE_PFAD): reset_alarm_historie() if not os.path.exists(QUITTIERUNGEN_PFAD): reset_quittierungen() if not os.path.exists(SIMULATION_STATE_PFAD): speichere_simulation_state("standard", "automatik") if not os.path.exists(ACTIVE_ALARM_STATE_PFAD): reset_active_alarm_state() if not os.path.exists(REPARATUR_HISTORIE_PFAD): reset_reparatur_historie() else: stelle_reparatur_historie_schema_sicher() if not os.path.exists(DATEN_PFAD): speichere_dashboard_daten([]) class WiederverwendbarerTCPServer(socketserver.TCPServer): """Erlaubt schnellen Neustart des lokalen Servers.""" allow_reuse_address = True class RuhigerRequestHandler(http.server.SimpleHTTPRequestHandler): """Behandelt Dashboard-Dateien und einfache Alarm-Quittierungen.""" def do_GET(self): parsed_url = urllib.parse.urlparse(self.path) if parsed_url.path == "/quittieren": self.handle_quittieren(parsed_url) return if parsed_url.path == "/kommentar": self.handle_kommentar(parsed_url) return if parsed_url.path == "/reparatur_speichern": self.handle_reparatur_speichern(parsed_url) return if parsed_url.path == "/simulation_setzen": self.handle_simulation_setzen(parsed_url) return super().do_GET() def handle_simulation_setzen(self, parsed_url): """Setzt den Demo-Simulationsmodus per Dashboard-Taster.""" parameter = urllib.parse.parse_qs(parsed_url.query) modus = parameter.get("modus", [None])[0] betriebsart = parameter.get("betriebsart", [None])[0] praesentation = parameter.get("praesentation", [None])[0] if praesentation: state, auto_quittieren = setze_praesentation(praesentation) else: if modus is not None and betriebsart is None: betriebsart = "wartung" if modus in ["reparatur_demo", "pruefprotokoll_erforderlich"] else "automatik" state = speichere_simulation_state(modus, betriebsart) auto_quittieren = False if state["anlagenmodus"] == "normal" and state["betriebsart"] == "automatik": reset_quittierungen() reset_active_alarm_state() # Die Demo-Taster sollen sofort sichtbare Wirkung haben. Darum wird # nach dem Umschalten direkt eine neue Messrunde geschrieben, statt auf # den naechsten periodischen Zyklus zu warten. fuehre_messrunde_aus() if auto_quittieren: quittiere_aktive_alarme_aus_dashboard() fuehre_messrunde_aus() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() antwort = json.dumps({ "ok": True, "modus": state["anlagenmodus"], "betriebsart": state["betriebsart"], "praesentation_aktiv": state.get("praesentation_aktiv", False), "praesentation_schritt": state.get("praesentation_schritt", 0), }, ensure_ascii=False) self.wfile.write(antwort.encode("utf-8")) def handle_quittieren(self, parsed_url): """Speichert eine Alarm-Quittierung per Browser-Klick.""" parameter = urllib.parse.parse_qs(parsed_url.query) alarm_typ = parameter.get("typ", ["motor"])[0] motorname = parameter.get("motor", [""])[0] sensor_id = parameter.get("id", [""])[0] daten = lese_vorherige_dashboard_daten() if alarm_typ == "sps": diagnose = next((eintrag for eintrag in daten.get("sps_io", {}).get("diagnosen", []) if eintrag.get("adresse") == sensor_id), None) if diagnose is None: self.send_response(404) self.end_headers() self.wfile.write(b"SPS-I/O-Alarm nicht gefunden") return quittierungen = lese_quittierungen() quittierungen[f"sps:{sensor_id}"] = { "status": diagnose["status"], "zeitstempel": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } speichere_quittierungen(quittierungen) speichere_ereignis( f"SPS {sensor_id}", "SPS I/O Foerderband 1", diagnose["status"], "SPS-I/O-Diagnose", 75 if diagnose["status"] == "Rot" else 45, f"SPS-I/O-Alarm {sensor_id} wurde quittiert", True, ) for eintrag in daten.get("sps_io", {}).get("diagnosen", []): if eintrag.get("adresse") == sensor_id: eintrag["alarm_quittiert"] = True for alarm in daten.get("aktive_alarme", []): if alarm.get("typ") == "sps" and alarm.get("id") == sensor_id: alarm["alarm_quittiert"] = True daten["ereignisprotokoll"] = lese_letzte_ereignisse() schreibe_json_datei(DATEN_PFAD, daten) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(b'{"ok": true}') return if alarm_typ == "sensor": sensor = next((eintrag for eintrag in daten.get("sensoren", []) if eintrag.get("id") == sensor_id), None) if sensor is None or sensor.get("status") == "OK": self.send_response(404) self.end_headers() self.wfile.write(b"Sensoralarm nicht gefunden") return quittierungen = lese_quittierungen() quittierungen[f"sensor:{sensor_id}"] = { "status": sensor["status"], "zeitstempel": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } speichere_quittierungen(quittierungen) speichere_ereignis( sensor["sensorname"], sensor["bereich"], "Rot" if sensor["status"] == "Fehler" else "Gelb", sensor["meldungstyp"], 60 if sensor["status"] == "Fehler" else 35, f"Sensoralarm {sensor['sensorname']} wurde quittiert", True, ) for eintrag in daten.get("sensoren", []): if eintrag.get("id") == sensor_id: eintrag["alarm_quittiert"] = True for alarm in daten.get("aktive_alarme", []): if alarm.get("typ") == "sensor" and alarm.get("id") == sensor_id: alarm["alarm_quittiert"] = True daten["ereignisprotokoll"] = lese_letzte_ereignisse() schreibe_json_datei(DATEN_PFAD, daten) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(b'{"ok": true}') return motor = None for eintrag in daten.get("motoren", []): if eintrag["motorname"] == motorname: motor = eintrag break if motor is None or motor.get("gesamtstatus") == "Gruen": self.send_response(404) self.end_headers() self.wfile.write(b"Alarm nicht gefunden") return quittierungen = lese_quittierungen() quittierungen[f"motor:{motorname}"] = { "status": motor["gesamtstatus"], "zeitstempel": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } speichere_quittierungen(quittierungen) speichere_ereignis( motor["motorname"], motor["bereich"], motor["gesamtstatus"], motor["meldungstyp"], motor["risikowert"], f"Alarm von {motor['motorname']} wurde quittiert", True, ) for eintrag in daten.get("motoren", []): if eintrag["motorname"] == motorname: eintrag["alarm_quittiert"] = True for alarm in daten.get("aktive_alarme", []): if alarm["motorname"] == motorname: alarm["alarm_quittiert"] = True daten["ereignisprotokoll"] = lese_letzte_ereignisse() schreibe_json_datei(DATEN_PFAD, daten) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(b'{"ok": true}') def handle_kommentar(self, parsed_url): """Speichert einen Bediener-Kommentar als Ereignis.""" parameter = urllib.parse.parse_qs(parsed_url.query) motorname = parameter.get("motor", [""])[0] kommentar = parameter.get("text", [""])[0].strip() daten = lese_vorherige_dashboard_daten() motor = None if not kommentar: self.send_response(400) self.end_headers() self.wfile.write(b"Kommentar fehlt") return for eintrag in daten.get("motoren", []): if eintrag["motorname"] == motorname: motor = eintrag break if motor is None: self.send_response(404) self.end_headers() self.wfile.write(b"Motor nicht gefunden") return speichere_ereignis( motor["motorname"], motor["bereich"], motor["gesamtstatus"], motor["meldungstyp"], motor["risikowert"], f"Bediener-Kommentar: {kommentar}", motor.get("alarm_quittiert", False), ) daten["ereignisprotokoll"] = lese_letzte_ereignisse() schreibe_json_datei(DATEN_PFAD, daten) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(b'{"ok": true}') def handle_reparatur_speichern(self, parsed_url): """Speichert das Reparaturformular aus dem Dashboard.""" parameter = urllib.parse.parse_qs(parsed_url.query) motorname = parameter.get("motor", [""])[0] schicht = parameter.get("schicht", [""])[0] fehlerart = parameter.get("fehlerart", [""])[0] mechanisches_fehlerbild = parameter.get("mechanisches_fehlerbild", [""])[0] elektrisches_fehlerbild = parameter.get("elektrisches_fehlerbild", [""])[0] mechanische_massnahme = parameter.get("mechanische_massnahme", [""])[0] elektrische_massnahme = parameter.get("elektrische_massnahme", [""])[0] teile_gewechselt = ", ".join(parameter.get("teile_gewechselt", [])) if mechanisches_fehlerbild == "-": mechanisches_fehlerbild = "" if elektrisches_fehlerbild == "-": elektrisches_fehlerbild = "" if mechanische_massnahme == "-": mechanische_massnahme = "" if elektrische_massnahme == "-": elektrische_massnahme = "" fehlerbild = parameter.get("fehlerbild", [mechanisches_fehlerbild or elektrisches_fehlerbild or fehlerart])[0] massnahme = parameter.get("massnahme", [mechanische_massnahme or elektrische_massnahme or "Keine Massnahme angegeben"])[0] ergebnis = parameter.get("ergebnis", [""])[0] kommentar = parameter.get("kommentar", [""])[0] mitarbeiter = parameter.get("mitarbeiter", [""])[0] or "Unbekannt" daten = lese_vorherige_dashboard_daten() motor = None for eintrag in daten.get("motoren", []): if eintrag["motorname"] == motorname: motor = eintrag break if motor is None: self.send_response(404) self.end_headers() self.wfile.write(b"Motor nicht gefunden") return speichere_reparatur_eintrag( motor, fehlerbild, massnahme, ergebnis, kommentar, mitarbeiter, schicht, fehlerart, mechanisches_fehlerbild, elektrisches_fehlerbild, mechanische_massnahme, elektrische_massnahme, teile_gewechselt, ) daten["reparaturhistorie"] = lese_letzte_reparaturen() daten["ereignisprotokoll"] = lese_letzte_ereignisse() schreibe_json_datei(DATEN_PFAD, daten) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(b'{"ok": true}') def log_message(self, format, *args): return def starte_lokalen_webserver(): """Startet den lokalen Webserver im Skriptordner.""" os.chdir(SCRIPT_ORDNER) handler = RuhigerRequestHandler try: server = WiederverwendbarerTCPServer(("", SERVER_PORT), handler) except OSError: print(f"Port {SERVER_PORT} ist bereits belegt.") print("Bitte schliesse das andere Programm oder stoppe den alten Python-Prozess.") return None thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server def fuehre_messrunde_aus(): """Simuliert eine Messrunde ohne parallele Schreibzugriffe.""" with MESSRUNDEN_LOCK: _fuehre_messrunde_aus_gesperrt() def _fuehre_messrunde_aus_gesperrt(): """Simuliert alle Motoren, bewertet sie und schreibt Dateien.""" historie = lese_historie() vorherige_daten = lese_vorherige_dashboard_daten() simulation_state = lade_simulation_state() berichte = [] for motor in MOTOREN: aktuelle_werte = simuliere_messwerte(motor, simulation_state) bericht = werte_motor_aus(motor, aktuelle_werte, historie, simulation_state) berichte.append(bericht) quittierungen = aktualisiere_quittierungen(berichte) sensoren = simuliere_zusatzsensoren(berichte, quittierungen, simulation_state) quittierungen = aktualisiere_sensor_quittierungen(sensoren, quittierungen) for sensor in sensoren: sensor["alarm_quittiert"] = ist_sensor_quittiert(sensor, quittierungen) sps_io = erstelle_sps_io(berichte, sensoren, quittierungen, simulation_state) quittierungen = aktualisiere_sps_quittierungen(sps_io, quittierungen) sps_io = erstelle_sps_io(berichte, sensoren, quittierungen, simulation_state) energieversorgung = erstelle_energieversorgung(simulation_state, berichte, sensoren) schrittkette = erstelle_schrittkette(berichte, sensoren, sps_io, simulation_state, energieversorgung) for bericht in berichte: bericht["alarm_quittiert"] = ist_alarm_quittiert(bericht, quittierungen) speichere_ereignisse_fuer_messrunde(berichte, vorherige_daten) for bericht in berichte: drucke_bericht(bericht) exportiere_wartungsbericht(bericht) speichere_historie(berichte) speichere_dashboard_daten(berichte, sensoren, sps_io, schrittkette) def main(): """Startet Dashboard, Webserver und Dauersimulation.""" os.chdir(SCRIPT_ORDNER) erstelle_projektdateien() if "--reset" in sys.argv: reset_historie() reset_alarm_historie() reset_quittierungen() reset_active_alarm_state() reset_reparatur_historie() speichere_simulation_state("standard", "automatik") print("motor_history.csv wurde zurueckgesetzt.") print("alarm_history.csv und acknowledged_alarms.json wurden zurueckgesetzt.") print("active_alarm_state.json wurde zurueckgesetzt.") print("repair_history.csv wurde zurueckgesetzt.") print("simulation_state.json wurde auf Standard-Demo gesetzt.") fuehre_messrunde_aus() server = starte_lokalen_webserver() if server is None: return print(f"Dashboard wird geoeffnet: {DASHBOARD_URL}") webbrowser.open(DASHBOARD_URL) try: while True: time.sleep(AKTUALISIERUNG_SEKUNDEN) fuehre_messrunde_aus() except KeyboardInterrupt: print("\nProgramm wird beendet.") server.shutdown() server.server_close() if __name__ == "__main__": main()