m.stangl/kabellaengen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ca2831ea2359eb4035d3f293f3580c5eae9dff3b
kabellaengen/lib/portalexport.py
T
lertlmaier ca2831ea23 In TIA-Export und WSCAD-Export den Kommentar zu fehlerhaften BMK's angepasst, sodass BMK genannt wird. 
Ausgelagerte Funktion für Duplikate und Error-Sheet erstellt und bei allen Export-Funktionen eingebaut
2025-07-04 09:19:19 +02:00

1269 lines
65 KiB
Python
Raw Blame History

import os
import pandas as pd # Required by ExcelConverter
import unittest
import pandas.testing as pdt
# Assuming the ExcelConverter class from the previous turn is available and pasted here
class ExcelConverter:
def __init__(self):
self.eingaenge_identifier = ["FDI", "B", "S", "E", "I", "DI", "BA", "BB", "BC", "BD", "BE", "BF", "BG", "BH", "BJ", "BK", "BL", "BM", "BN", "BP", "BQ", "BR", "BS", "BT", "BU", "BV", "BW", "BX", "BY", "BZ", "FA", "FB", "FC", "FD", "FE", "FF", "FG", "FH", "FJ", "FK", "FL", "FM", "FN", "FP", "FQ", "FR", "FS", "FT", "FU", "FV", "FW", "FX", "FY", "FZ", "SA", "SB", "SC", "SD", "SE", "SF", "SG", "SH", "SJ", "SK", "SL", "SM", "SN", "SP", "SQ", "SR", "SS", "ST", "SU", "SV", "SW", "SX", "SY", "SZ"]
self.ausgaenge_identifier = ["K", "O", "Q", "M", "A", "P", "DQ", "CA", "CB", "CC", "CD", "CE", "CF", "CG", "CH", "CJ", "CK", "CL", "CM", "CN", "CP", "CQ", "CR", "CS", "CT", "CU", "CV", "CW", "CX", "CY", "CZ", "EA", "EB", "EC", "ED", "EE", "EF", "EG", "EH", "EJ", "EK", "EL", "EM", "EN", "EP", "EQ", "ER", "ES", "ET", "EU", "EV", "EW", "EX", "EY", "EZ", "GA", "GB", "GC", "GD", "GE", "GF", "GG", "GH", "GJ", "GK", "GL", "GM", "GN", "GP", "GQ", "GR", "GS", "GT", "GU", "GV", "GW", "GX", "GY", "GZ", "HA", "HB", "HC", "HD", "HE", "HF", "HG", "HH", "HJ", "HK", "HL", "HM", "HN", "HP", "HQ", "HR", "HS", "HT", "HU", "HV", "HW", "HX", "HY", "HZ", "KA", "KB", "KC", "KD", "KE", "KF", "KG", "KH", "KJ", "KK", "KL", "KM", "KN", "KP", "KQ", "KR", "KS", "KT", "KU", "KV", "KW", "KX", "KY", "KZ", "MA", "MB", "MC", "MD", "ME", "MF", "MG", "MH", "MJ", "MK", "ML", "MM", "MN", "MP", "MQ", "MR", "MQ", "MR", "MS", "MT", "MU", "MV", "MW", "MX", "MY", "MZ", "PA", "PB", "PC", "PD", "PE", "PF", "PG", "PH", "PJ", "PK", "PL", "PM", "PN", "PP", "PQ", "PR", "PS", "PT", "PU", "PV", "PW", "PX", "PY", "PZ", "QA", "QB", "QC", "QD", "QE", "QF", "QG", "QH", "QJ", "QK", "QL", "QM", "QN", "QP", "QQ", "QR", "QS", "QT", "QU", "QV", "QW", "QX", "QY", "QZ", "RA", "RB", "RC", "RD", "RE", "RF", "RG", "RH", "RJ", "RK", "RL", "RM", "RN", "RP", "RQ", "RR", "RS", "RT", "RU", "RV", "RW", "RX", "RY", "RZ", "TA", "TB", "TC", "TD", "TE", "TF", "TG", "TH", "TJ", "TK", "TL", "TM", "TN", "TP", "TQ", "TR", "TS", "TT", "TU", "TV", "TW", "TX", "TY", "TZ"]
def _excel_datei_name_erstellen(self, eingestellter_name, pfad_ursprungsdatei, pfad_ziel_datei):
"""
Erstellt den vollständigen Dateinamen für die Excel-Zieldatei.
Entspricht der PowerShell-Funktion ExcelDateiNameErstellen.
"""
name_ursprungsdatei = os.path.basename(pfad_ursprungsdatei)
name_ursprungsdatei_kurz = os.path.splitext(name_ursprungsdatei)[0]
gesamt_pfad_ziel_datei = os.path.join(pfad_ziel_datei, eingestellter_name.replace("[Ursprungsdatei]", name_ursprungsdatei_kurz) + ".xlsx")
return gesamt_pfad_ziel_datei
def _gleiche_finden_old(self, zu_ueberpruefen):
"""
Findet doppelte Einträge basierend auf dem 'Name'-Feld.
Entspricht der PowerShell-Funktion GleicheFinden.
"""
zu_ueberpruefen_sorted = sorted(zu_ueberpruefen, key=lambda x: x['Name'])
gleiche = []
for i in range(len(zu_ueberpruefen_sorted) - 1):
if zu_ueberpruefen_sorted[i]['Name'] == zu_ueberpruefen_sorted[i+1]['Name']:
# Füge Eintrag hinzu, wenn er noch nicht hinzugefügt wurde, um Duplikate im 'gleiche' Array zu vermeiden
if not gleiche or gleiche[-1]['Zeile'] != zu_ueberpruefen_sorted[i]['Zeile']:
gleiche.append(zu_ueberpruefen_sorted[i])
gleiche.append(zu_ueberpruefen_sorted[i+1])
return gleiche
def _gleiche_finden(self, liste):
counter = {}
for eintrag in liste:
name = eintrag['Name']
counter[name] = counter.get(name, 0) + 1
doppelte = set(name for name, count in counter.items() if count > 1)
doppelte_eintraege = [eintrag for eintrag in liste if eintrag['Name'] in doppelte]
return doppelte_eintraege, doppelte
def _anschluesse_ergaenzen_old(self, anschluesse):
"""
Ergänzt fehlende Anschlüsse für Bauteile.
Entspricht der PowerShell-Funktion AnschluesseErgaenzen.
"""
anschluesse_sorted = sorted(anschluesse, key=lambda x: x['Name'])
# Erstelle eine Kopie der Liste, um Modifikationen während der Iteration zu vermeiden
anschluesse_copy = list(anschluesse_sorted)
added_entries = []
for anschluss in anschluesse_copy:
anschluss_bauteil = anschluss['Name'][:-2]
soll_anschluesse = set(range(8)) # 0 bis 7
for sub_anschluss in anschluesse_copy + added_entries: # Berücksichtige bereits hinzugefügte Einträge
sub_anschluss_bauteil = sub_anschluss['Name'][:-2]
if sub_anschluss_bauteil == anschluss_bauteil:
sub_anschluss_letzte_stelle = int(sub_anschluss['Name'][-1])
if sub_anschluss_letzte_stelle in soll_anschluesse:
soll_anschluesse.remove(sub_anschluss_letzte_stelle)
for fehlender_anschluss in soll_anschluesse:
new_name = anschluss_bauteil + str(fehlender_anschluss)
new_entry = {
'Name': new_name,
'Kommentar': "",
'Bezug': "" if 'Bezug' in anschluss else "", # Füge 'Bezug' hinzu, falls es in der ursprünglichen Struktur vorhanden war
'Zeile': -1 # Von Loernz hinzugefügt da sonst code crasht, wenn später in _gleiche_finden über 'Zeile' iteriert wird
}
added_entries.append(new_entry)
anschluesse.extend(added_entries)
return sorted(anschluesse, key=lambda x: x['Name'])
def _anschluesse_ergaenzen(self, anschluesse):
"""
Ergänzt fehlende Anschlüsse für Bauteile.
Entspricht der PowerShell-Funktion AnschluesseErgaenzen.
"""
anschluesse_sorted = sorted(anschluesse, key=lambda x: x['Name'])
bauteil_dict = {}
# Gruppiere vorhandene Anschlüsse nach Bauteil
for eintrag in anschluesse_sorted:
name = eintrag['Name']
if '.' in name:
bauteil = name.rsplit('.', 1)[0] + '.'
nummer = int(name.split('.')[-1])
else:
bauteil = name[:-1]
nummer = int(name[-1])
if bauteil not in bauteil_dict:
bauteil_dict[bauteil] = set()
bauteil_dict[bauteil].add(nummer)
added_entries = []
for bauteil, vorhandene_nummern in bauteil_dict.items():
fehlende = set(range(8)) - vorhandene_nummern
for i in fehlende:
new_name = f"{bauteil}{i}"
new_entry = {
'Name': new_name,
'Kommentar': "",
'Bezug': "",
'Zeile': -1
}
added_entries.append(new_entry)
anschluesse.extend(added_entries)
return sorted(anschluesse, key=lambda x: x['Name'])
def _verarbeite_io_liste(self, liste):
"""
Filtert doppelte Namen aus der Liste und markiert sie mit Hinweis; gibt separate Liste mit eindeutigen Duplikaten zurück.
"""
gleiche, doppelte_namen = self._gleiche_finden(liste)
# Nur ein Eintrag pro doppeltem Namen in gleiche_liste
gesehen = set()
reduzierte_gleiche = []
for eintrag in gleiche:
name = eintrag['Name']
if name not in gesehen:
reduzierte_gleiche.append(eintrag)
gesehen.add(name)
gefiltert = self._duplikate_markieren_und_filtern(liste, doppelte_namen)
return reduzierte_gleiche, gefiltert
def _duplikate_markieren_und_filtern(self, liste, doppelte_namen):
"""
Behalte nur das erste Vorkommen jedes doppelten Namens in der Liste.
Markiere dieses Vorkommen mit 'Hinweis': 'Doppelt'.
"""
neue_liste = []
gesehen = set()
for eintrag in liste:
name = eintrag['Name']
if name in doppelte_namen:
if name not in gesehen:
eintrag['Hinweis'] = "Duplikat!"
neue_liste.append(eintrag)
gesehen.add(name)
# sonst: ignorieren
else:
neue_liste.append(eintrag)
return neue_liste
def _prep_simpel(self, ursprungdatei, prefix_eingang="E", prefix_ausgang="A"):
"""
Konvertiert Daten aus der Ursprungsdatei
Eingänge und Ausgänge in seperaten dataframes; Fehler nicht enthalten
"""
eingaenge = []
ausgaenge = []
fehler = []
for i, eintrag in enumerate(ursprungdatei):
eintragszeile = eintrag.strip().split("'")
bmk_zahl_raw = eintragszeile[1].strip()
kommentar = eintragszeile[3].strip()
bmk_alpha = ''.join(filter(str.isalpha, bmk_zahl_raw))
bmk_numeric = ''.join(filter(str.isdigit, bmk_zahl_raw))
bmk_formatted = bmk_numeric
if len(bmk_numeric) > 0:
bmk_formatted = bmk_numeric[:-1] + "." + bmk_numeric[-1] if len(bmk_numeric) > 1 else "0." + bmk_numeric
if bmk_alpha in self.eingaenge_identifier:
name = prefix_eingang + bmk_formatted
eingaenge.append({'Name': name, 'Kommentar': kommentar, 'Zeile': i})
elif bmk_alpha in self.ausgaenge_identifier:
name = prefix_ausgang + bmk_formatted
ausgaenge.append({'Name': name, 'Kommentar': kommentar, 'Zeile': i})
else:
name = bmk_formatted
fehler.append({'Name': name, 'Kommentar': f"Fehlerhafte BMK '{bmk_zahl_raw}'", 'Zeile': i})
# Fehler-Dataframe erzeugen
df_fehler = pd.DataFrame(fehler).sort_values(by="Name") if fehler else pd.DataFrame()
# Eingänge & Ausgänge verarbeiten (Duplikate markieren und filtern)
gleiche_eingaenge, eingaenge = self._verarbeite_io_liste(eingaenge)
gleiche_ausgaenge, ausgaenge = self._verarbeite_io_liste(ausgaenge)
# DataFrames erzeugen u. sortieren
df_eingaenge = pd.DataFrame(eingaenge).sort_values(by="Name")
df_ausgaenge = pd.DataFrame(ausgaenge).sort_values(by="Name")
df_gleiche_eingaenge = pd.DataFrame(gleiche_eingaenge).sort_values(by="Name") if gleiche_eingaenge else pd.DataFrame()
df_gleiche_ausgaenge = pd.DataFrame(gleiche_ausgaenge).sort_values(by="Name") if gleiche_ausgaenge else pd.DataFrame()
return df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge
def _prep_tia(self, ursprungdatei, prefix_eingang="E", prefix_ausgang="A"):
"""
Konvertiert Daten für TIA Portal und schreibt sie in eine Excel-Datei mit zwei Blättern ("Constants" und "PLC Tags").
Eingänge und Ausgänge gemeinsam jeweils in consts und tags (Prefix cIn, cOut); doppelte Namen markiert und ausgeschlossen; Fehler mit Prefix 'Err'.
"""
eingaenge = []
ausgaenge = []
fehler = []
for i, eintrag in enumerate(ursprungdatei):
eintragszeile = eintrag.strip().split("'")
bmk_zahl_raw = eintragszeile[1].strip()
kommentar = eintragszeile[3].strip()
bmk_alpha = ''.join(filter(str.isalpha, bmk_zahl_raw))
bmk_numeric = ''.join(filter(str.isdigit, bmk_zahl_raw))
bmk_formatted = bmk_numeric
if len(bmk_numeric) > 0:
bmk_formatted = bmk_numeric[:-1] + "." + bmk_numeric[-1] if len(bmk_numeric) > 1 else "0." + bmk_numeric
if bmk_alpha in self.eingaenge_identifier:
name = prefix_eingang + bmk_formatted
name_constant = "cIn" + bmk_zahl_raw
eingaenge.append({
'Name': name,
'NameOriginal': bmk_zahl_raw,
'NameConstant': name_constant,
'Kommentar': kommentar,
'Zahl': int(bmk_numeric) if bmk_numeric else 0, # Convert to int
'Zeile': i
})
elif bmk_alpha in self.ausgaenge_identifier:
name = prefix_ausgang + bmk_formatted
name_constant = "cOut" + bmk_zahl_raw
ausgaenge.append({
'Name': name,
'NameOriginal': bmk_zahl_raw,
'NameConstant': name_constant,
'Kommentar': kommentar,
'Zahl': int(bmk_numeric) if bmk_numeric else 0, # Convert to int
'Zeile': i
})
else: # Hier wird Fehlerhafte BMK behandelt
name = bmk_formatted
name_constant = "cErr" + bmk_zahl_raw
fehler.append({
'Name': name,
'NameOriginal': bmk_zahl_raw,
'NameConstant': name_constant,
'Kommentar': f"Fehlerhafte BMK '{bmk_zahl_raw}'", # Zeile von Lorenz eingefügt, da sonst pandas mit Fehler, da kein Kommentar
'Zahl': i, # Original PowerShell uses 'i' for Zahl in Fehler, not bmk_numeric
'Zeile': i
})
# Fehler-Dataframe erzeugen
df_fehler = pd.DataFrame(fehler).sort_values(by="Zahl") if fehler else pd.DataFrame()
# Duplikate verarbeiten & markieren
gleiche_eingaenge, eingaenge = self._verarbeite_io_liste(eingaenge)
gleiche_ausgaenge, ausgaenge = self._verarbeite_io_liste(ausgaenge)
# DataFrames erstellen u. sortieren
df_eingaenge = pd.DataFrame(eingaenge).sort_values(by="Zahl")
df_ausgaenge = pd.DataFrame(ausgaenge).sort_values(by="Zahl")
df_gleiche_eingaenge = pd.DataFrame(gleiche_eingaenge).sort_values(by="Zahl") if gleiche_eingaenge else pd.DataFrame()
df_gleiche_ausgaenge = pd.DataFrame(gleiche_ausgaenge).sort_values(by="Zahl") if gleiche_ausgaenge else pd.DataFrame()
df_consts, df_tags = self._create_tia_dataframes(df_fehler, df_eingaenge, df_ausgaenge)
return df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_consts, df_tags
def _create_tia_dataframes(self, df_fehler, df_eingaenge, df_ausgaenge):
"""
Hilfsfunktion für 'prep_tia', die dataframes für spätere excel Ausgabe erzeugt ("Constants" und "PLC Tags").
Eingänge und Ausgänge gemeinsam jeweils in consts und tags; Fehler ebenfalls mit Prefix 'Err' enthalten.
"""
df_constants_data = []
j = 0
for row in df_fehler.itertuples():
df_constants_data.append({
"Name": row.NameConstant,
"Path": "Import",
"Data Type": "Int",
"Value": j,
"Comment": ""
})
j += 1
for row in df_eingaenge.itertuples():
df_constants_data.append({
"Name": row.NameConstant,
"Path": "Import",
"Data Type": "Int",
"Value": j,
"Comment": ""
})
j += 1
for row in df_ausgaenge.itertuples():
df_constants_data.append({
"Name": row.NameConstant,
"Path": "Import",
"Data Type": "Int",
"Value": j,
"Comment": ""
})
j += 1
df_constants = pd.DataFrame(df_constants_data)
# Blatt "PLC Tags"
df_plc_tags_data = []
for row in df_fehler.itertuples():
df_plc_tags_data.append({
"Name": row.NameOriginal,
"Path": "Import",
"Data Type": "Bool",
"Logical Address": row.Name,
"Comment": row.Kommentar,
"Hmi Visible": "'True",
"Hmi Accessible": "'True",
"Hmi Writeable": "'True",
"Typeobject ID": "",
"Version ID": ""
})
for row in df_eingaenge.itertuples() :
df_plc_tags_data.append({
"Name": row.NameOriginal,
"Path": "Import",
"Data Type": "Bool",
"Logical Address": row.Name,
"Comment": row.Kommentar,
"Hmi Visible": "'True",
"Hmi Accessible": "'True",
"Hmi Writeable": "'True",
"Typeobject ID": "",
"Version ID": ""
})
for row in df_ausgaenge.itertuples():
df_plc_tags_data.append({
"Name": row.NameOriginal,
"Path": "Import",
"Data Type": "Bool",
"Logical Address": row.Name,
"Comment": row.Kommentar,
"Hmi Visible": "'True",
"Hmi Accessible": "'True",
"Hmi Writeable": "'True",
"Typeobject ID": "",
"Version ID": ""
})
df_plc_tags = pd.DataFrame(df_plc_tags_data)
return (df_constants, df_plc_tags)
def _prep_wscad(self, ursprungdatei, prefix_eingang="E", prefix_ausgang="A"):
"""
Konvertiert Daten für WSCAD mit Bezugsinformationen und schreibt sie in eine Excel-Datei.
Entspricht der PowerShell-Funktion ExcelDateiBeschreibenWSCAD_Bezug.
"""
eingaenge = []
ausgaenge = []
fehler = []
for i, eintrag in enumerate(ursprungdatei):
eintragszeile = eintrag.strip().split("'")
bmk_zahl_raw = eintragszeile[1].strip()
kommentar = eintragszeile[3].strip()
bezug = eintragszeile[7].strip()
bmk_alpha = ''.join(filter(str.isalpha, bmk_zahl_raw))
bmk_numeric = ''.join(filter(str.isdigit, bmk_zahl_raw))
bmk_formatted = bmk_numeric
if len(bmk_numeric) > 0:
bmk_formatted = bmk_numeric[:-1] + "." + bmk_numeric[-1] if len(bmk_numeric) > 1 else "0." + bmk_numeric
# Remove leading zeros
if len(bmk_formatted) >= 4 and bmk_formatted.startswith("0") and bmk_alpha in self.eingaenge_identifier:
bmk_formatted = bmk_formatted[1:]
if len(bmk_formatted) >= 4 and bmk_formatted.startswith("0"):
bmk_formatted = bmk_formatted[1:]
elif len(bmk_formatted) >= 3 and bmk_formatted.startswith("0") and bmk_alpha in self.ausgaenge_identifier:
bmk_formatted = bmk_formatted[1:]
if len(bmk_formatted) >= 3 and bmk_formatted.startswith("0"):
bmk_formatted = bmk_formatted[1:]
if bmk_alpha in self.eingaenge_identifier:
name = prefix_eingang + bmk_formatted
eingaenge.append({'Name': name, 'Kommentar': kommentar, 'Zeile': i, 'Bezug': bezug})
elif bmk_alpha in self.ausgaenge_identifier:
name = prefix_ausgang + bmk_formatted
ausgaenge.append({'Name': name, 'Kommentar': kommentar, 'Zeile': i, 'Bezug': bezug})
else:
name = bmk_formatted
fehler.append({'Name': name, 'Kommentar': f"Fehlerhafte BMK '{bmk_zahl_raw}'", 'Zeile': i, 'Bezug': bezug})
eingaenge = self._anschluesse_ergaenzen(eingaenge)
ausgaenge = self._anschluesse_ergaenzen(ausgaenge)
# Fehler-Dataframe erzeugen
df_fehler = pd.DataFrame(fehler).sort_values(by="Name") if fehler else pd.DataFrame()
# Duplikate verarbeiten & markieren
gleiche_eingaenge, eingaenge = self._verarbeite_io_liste(eingaenge)
gleiche_ausgaenge, ausgaenge = self._verarbeite_io_liste(ausgaenge)
# DataFrames erzeugen u. sortieren
df_eingaenge = pd.DataFrame(eingaenge).sort_values(by="Name")
df_ausgaenge = pd.DataFrame(ausgaenge).sort_values(by="Name")
df_gleiche_eingaenge = pd.DataFrame(gleiche_eingaenge).sort_values(by="Name") if gleiche_eingaenge else pd.DataFrame()
df_gleiche_ausgaenge = pd.DataFrame(gleiche_ausgaenge).sort_values(by="Name") if gleiche_ausgaenge else pd.DataFrame()
return df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge
def _prep_data(self, data, export_type):
"""
Aufbereiter der Daten je nach gewünschtem Export-Typ
"""
if export_type == "Simpel":
return self._prep_simpel(data)
elif export_type == "TIA":
return self._prep_tia(data)
elif export_type == "WSCAD" or "WSCAD mit Bezug":
return self._prep_wscad(data)
else:
raise ValueError("Unbekannter Export-Typ")
def export_excel(self, data, export_type, export_path):
"""
Wrapper, der aufgerufen wird und in sich die Aufbereitung der Daten und das Abspeichern als Excel abruft.
"""
# Daten vorbereiten
dataframes = self._prep_data(data, export_type)
# Excel Datei schreiben und speichern
self._write_excel(dataframes, export_type, export_path)
def _write_excel(self, dataframes, export_type, export_path):
"""
Erzeugung der Excel-Datei im gewünschten Muster
"""
if export_type == "Simpel":
# Format für Simpel-Ausgabe
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge = dataframes
with pd.ExcelWriter(export_path, engine='openpyxl') as writer:
# Sheet 1: Ein / Ausgänge
start_row = 0
df_eingaenge.to_excel(writer, sheet_name='Ein-, Ausgaenge', index=False, startrow=start_row)
start_row += len(df_eingaenge) + 2 # Leerzeile dazwischen
# Ausgaenge direkt drunter schreiben (ohne Header)
df_ausgaenge.to_excel(writer, sheet_name='Ein-, Ausgaenge', index=False, startrow=start_row, header=False)
# Duplikate und Errors
self._write_duplicates_errors(writer, df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler)
elif export_type == "TIA":
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_consts, df_tags = dataframes
with pd.ExcelWriter(export_path, engine='openpyxl') as writer:
# 1. Constants Sheet mit Header. Spalten: Name, Path, Data Type, Value, Comment
df_consts.to_excel(writer, sheet_name='Constants', index=False)
# 2. PLC Tags Sheet mit Header. Spalten: Name, Path, Data Type, Logical Address, Comment, Hmi Visible, Hmi Accessible, Hmi Writeable, Typeobject ID, Version ID
df_tags.to_excel(writer, sheet_name='PLC Tags', index=False)
# Duplikate und Errors
self._write_duplicates_errors(writer, df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler)
elif export_type == "WSCAD":
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge = dataframes
# Basis-Template
df_result = pd.DataFrame(
columns=[
"Anschluss",
"Kommentar",
"Symboltext",
"Bezug",
"Sondertext 1",
"Sondertext 2",
"Sondertext 3",
"Sondertext 4",
"Sondertext 5",
"Sondertext 6",
"Sondertext 7",
"Sondertext 8",
]
)
df_result = self._append_df(df_result, df_fehler)
df_result = self._append_df(df_result, df_eingaenge)
df_result = self._append_df(df_result, df_ausgaenge)
with pd.ExcelWriter(export_path, engine="openpyxl") as writer:
df_result.to_excel(writer, index=False, sheet_name="WSCAD")
# Duplikate und Errors
self._write_duplicates_errors(writer, df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler)
elif export_type == "WSCAD mit Bezug":
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge = dataframes
# Basis-Template
df_result = pd.DataFrame(
columns=[
"Anschluss",
"Kommentar",
"Symboltext",
"Bezug",
"Sondertext 1",
"Sondertext 2",
"Sondertext 3",
"Sondertext 4",
"Sondertext 5",
"Sondertext 6",
"Sondertext 7",
"Sondertext 8",
]
)
df_result = self._append_df(df_result, df_fehler, include_bezug=True)
df_result = self._append_df(df_result, df_eingaenge, include_bezug=True)
df_result = self._append_df(df_result, df_ausgaenge, include_bezug=True)
with pd.ExcelWriter(export_path, engine="openpyxl") as writer:
df_result.to_excel(writer, index=False, sheet_name="WSCAD")
# Duplikate und Errors
self._write_duplicates_errors(writer, df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler)
def _append_df(self, target_df, source_df, include_bezug=False):
"""
Hilfsfunktion, die Dataframes für WSCAD Export aneinander reiht und entweder Bezug inkludiert oder nicht
"""
data = {
"Anschluss": source_df["Name"],
"Kommentar": source_df["Kommentar"],
"Symboltext": "",
"Bezug": source_df["Bezug"].apply(lambda x: f"'{x}" if pd.notna(x) else "") if include_bezug else "",
"Sondertext 1": "",
"Sondertext 2": "",
"Sondertext 3": "",
"Sondertext 4": "",
"Sondertext 5": "",
"Sondertext 6": "",
"Sondertext 7": "",
"Sondertext 8": "",
}
tmp = pd.DataFrame(data)
return pd.concat([target_df, tmp], ignore_index=True)
def _write_duplicates_errors(self, writer, df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler):
"""
Schreibt optional die Sheets 'Duplikate' und 'Fehler', falls Inhalte vorhanden sind.
"""
# Bezug-Spalte für Excel Darstellung vorbereiten
for df in [df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler]:
if "Bezug" in df.columns:
df["Bezug"] = df["Bezug"].apply(lambda x: f"'{x}" if pd.notnull(x) else x)
# Duplikate-Sheet
if not df_gleiche_eingaenge.empty or not df_gleiche_ausgaenge.empty:
start_row = 0
if not df_gleiche_eingaenge.empty:
df_gleiche_eingaenge.to_excel(writer, sheet_name='Duplikate', index=False, startrow=start_row)
start_row += len(df_gleiche_eingaenge) + 2 # Abstand
if not df_gleiche_ausgaenge.empty:
df_gleiche_ausgaenge.to_excel(writer, sheet_name='Duplikate', index=False, startrow=start_row, header=False)
# Fehler-Sheet
if not df_fehler.empty:
df_fehler.to_excel(writer, sheet_name='Fehler', index=False)
class IOConverterApp:
def __init__(self, master):
self.master = master
master.title("Dateikonvertierer I/O")
master.geometry("940x660")
master.resizable(False, False)
self.notebook = ttk.Notebook(master)
self.notebook.pack(padx=10, pady=10, expand=True, fill="both")
self.excel_converter = ExcelConverter()
self.selected_txt_file = ""
self.selected_target_directory = ""
self.target_file_configurations = [] # Stores configurations from settings tab
# Tab: TXT-Datei Auswaehlen
self.tab_txt_select = ttk.Frame(self.notebook)
self.notebook.add(self.tab_txt_select, text="txt-Datei Auswaehlen")
self.txt_path_field = ttk.Entry(self.tab_txt_select)
self.txt_path_field.place(x=10, y=10, width=860, height=20)
self.txt_path_field.bind("<Return>", self.txt_path_field_return)
self.txt_list_box = tk.Listbox(self.tab_txt_select, font=("Arial", 12))
self.txt_list_box.place(x=10, y=40, width=860, height=450)
self.txt_list_box.bind("<Double-Button-1>", self.txt_list_box_double_click)
self.txt_list_box.bind("<BackSpace>", self.txt_list_box_back_key)
self.txt_back_button = ttk.Button(self.tab_txt_select, text="Zurueck", command=self.txt_back_button_click)
self.txt_back_button.place(x=10, y=500, width=100, height=30)
self.txt_select_button = ttk.Button(self.tab_txt_select, text="Auswaehlen", command=self.txt_select_button_click)
self.txt_select_button.place(x=750, y=500, width=100, height=30)
# Tab: Zielverzeichnis Auswaehlen
self.tab_target_dir_select = ttk.Frame(self.notebook)
self.notebook.add(self.tab_target_dir_select, text="Zielverzeichnis Auswaehlen")
self.target_dir_path_field = ttk.Entry(self.tab_target_dir_select)
self.target_dir_path_field.place(x=10, y=10, width=860, height=20)
self.target_dir_path_field.bind("<Return>", self.target_dir_path_field_return)
self.target_dir_list_box = tk.Listbox(self.tab_target_dir_select, font=("Arial", 12))
self.target_dir_list_box.place(x=10, y=40, width=860, height=450)
self.target_dir_list_box.bind("<Double-Button-1>", self.target_dir_list_box_double_click)
self.target_dir_list_box.bind("<BackSpace>", self.target_dir_list_box_back_key)
self.target_dir_back_button = ttk.Button(self.tab_target_dir_select, text="Zurueck", command=self.target_dir_back_button_click)
self.target_dir_back_button.place(x=10, y=500, width=100, height=30)
self.target_dir_select_button = ttk.Button(self.tab_target_dir_select, text="Auswaehlen", command=self.target_dir_select_button_click)
self.target_dir_select_button.place(x=630, y=500, width=100, height=30)
self.target_dir_confirm_button = ttk.Button(self.tab_target_dir_select, text="Bestaetigen", command=self.target_dir_confirm_button_click)
self.target_dir_confirm_button.place(x=750, y=500, width=100, height=30)
# Tab: Bearbeitungsfortschritt
self.tab_progress = ttk.Frame(self.notebook)
self.notebook.add(self.tab_progress, text="Bearbeitungsfortschritt")
ttk.Label(self.tab_progress, text="Aktuelle Aufgabe").place(x=10, y=10, width=130, height=20)
self.current_task_label = ttk.Label(self.tab_progress, text="txt-Datei auswaehlen", font=("Arial", 13))
self.current_task_label.place(x=150, y=10, width=710, height=60)
ttk.Label(self.tab_progress, text="Erledigte Aufgaben").place(x=10, y=55, width=430, height=20)
self.completed_tasks_listbox = tk.Listbox(self.tab_progress, font=("Arial", 12))
self.completed_tasks_listbox.place(x=10, y=85, width=860, height=200)
ttk.Label(self.tab_progress, text="Fortschritt Excel-Dateien beschreiben").place(x=10, y=300, width=430, height=20)
self.progress_bar = ttk.Progressbar(self.tab_progress, orient="horizontal", mode="determinate")
self.progress_bar.place(x=10, y=320, width=860, height=30)
ttk.Label(self.tab_progress, text="Bearbeitete Eintraege:").place(x=10, y=370, width=140, height=20)
self.processed_entries_label = ttk.Label(self.tab_progress, text="0")
self.processed_entries_label.place(x=170, y=370, width=430, height=20)
ttk.Label(self.tab_progress, text="Gesamteintraege:").place(x=10, y=400, width=100, height=20)
self.total_entries_label = ttk.Label(self.tab_progress, text="0")
self.total_entries_label.place(x=170, y=400, width=430, height=20)
# Tab: Doppelte Ein- und Ausgaenge
self.tab_duplicates = ttk.Frame(self.notebook)
self.notebook.add(self.tab_duplicates, text="Doppelte Ein- und Ausgaenge")
ttk.Label(self.tab_duplicates, text="Bei den Eingaengen gibt es folgende Duplikate: ").place(x=10, y=10, width=275, height=40)
self.duplicate_inputs_treeview = ttk.Treeview(self.tab_duplicates, columns=("Zeile", "Kennzeichnung"), show="headings")
self.duplicate_inputs_treeview.heading("Zeile", text="Zeile")
self.duplicate_inputs_treeview.heading("Kennzeichnung", text="Kennzeichnung")
self.duplicate_inputs_treeview.column("Zeile", width=130, anchor="w")
self.duplicate_inputs_treeview.column("Kennzeichnung", width=130, anchor="w")
self.duplicate_inputs_treeview.place(x=10, y=60, width=275, height=430)
ttk.Label(self.tab_duplicates, text="Bei den Ausgaengen gibt es folgende Duplikate: ").place(x=295, y=10, width=275, height=40)
self.duplicate_outputs_treeview = ttk.Treeview(self.tab_duplicates, columns=("Zeile", "Kennzeichnung"), show="headings")
self.duplicate_outputs_treeview.heading("Zeile", text="Zeile")
self.duplicate_outputs_treeview.heading("Kennzeichnung", text="Kennzeichnung")
self.duplicate_outputs_treeview.column("Zeile", width=130, anchor="w")
self.duplicate_outputs_treeview.column("Kennzeichnung", width=130, anchor="w")
self.duplicate_outputs_treeview.place(x=295, y=60, width=275, height=430)
ttk.Label(self.tab_duplicates, text="Folgende Eintraege konnten keinem Betriebsmittel zugewiesen werden:").place(x=580, y=10, width=275, height=40)
self.unassigned_errors_treeview = ttk.Treeview(self.tab_duplicates, columns=("Zeile", "Kennzeichnung"), show="headings")
self.unassigned_errors_treeview.heading("Zeile", text="Zeile")
self.unassigned_errors_treeview.heading("Kennzeichnung", text="Kennzeichnung")
self.unassigned_errors_treeview.column("Zeile", width=130, anchor="w")
self.unassigned_errors_treeview.column("Kennzeichnung", width=130, anchor="w")
self.unassigned_errors_treeview.place(x=580, y=60, width=275, height=430)
self.exit_button = ttk.Button(self.tab_duplicates, text="Programm beenden", command=master.quit)
self.exit_button.place(x=10, y=500, width=860, height=30)
# Tab: Einstellungen
self.tab_settings = ttk.Frame(self.notebook)
self.notebook.add(self.tab_settings, text="Einstellungen")
ttk.Label(self.tab_settings, text="Startverzeichnis Auswahl .txt-Datei:").place(x=10, y=10, width=200, height=20)
self.txt_start_path_label = ttk.Label(self.tab_settings, text="")
self.txt_start_path_label.place(x=210, y=10, width=510, height=20)
ttk.Button(self.tab_settings, text="Aendern", command=self.change_txt_start_path).place(x=730, y=10, width=150, height=30)
ttk.Label(self.tab_settings, text="Startverzeichnis bei Auswahl des Zielverzeichnisses der Datei(en):").place(x=10, y=45, width=450, height=20)
self.target_path_option_var = tk.StringVar(value="WieTXT")
self.radio_button_frame = ttk.LabelFrame(self.tab_settings, text="")
self.radio_button_frame.place(x=10, y=60, width=449, height=30)
ttk.Radiobutton(self.radio_button_frame, text="Gleich Startpfad .txt-Datei", variable=self.target_path_option_var, value="WieTXT").place(x=10, y=10)
ttk.Radiobutton(self.radio_button_frame, text="Gleich gewaehlter .txt-Datei", variable=self.target_path_option_var, value="WieZielDatei").place(x=173, y=10)
ttk.Radiobutton(self.radio_button_frame, text="Anderer Pfad", variable=self.target_path_option_var, value="EigenerPfad").place(x=352, y=10)
ttk.Button(self.tab_settings, text="Speichern", command=self.save_radio_button_settings).place(x=730, y=60, width=150, height=30)
ttk.Label(self.tab_settings, text="Startverzeichnis Auswahl Zielverzeichnis:").place(x=10, y=95, width=250, height=20)
self.target_dir_start_path_label = ttk.Label(self.tab_settings, text="")
self.target_dir_start_path_label.place(x=260, y=95, width=460, height=20)
ttk.Button(self.tab_settings, text="Aendern", command=self.change_target_dir_start_path).place(x=730, y=95, width=150, height=30)
ttk.Label(self.tab_settings, text="Zieldateien:").place(x=10, y=140, width=200, height=20)
ttk.Button(self.tab_settings, text="Aenderungen Speichern", command=self.save_target_file_settings).place(x=730, y=170, width=150, height=30)
self.target_files_treeview = ttk.Treeview(self.tab_settings, columns=("Prefix Eingang", "Prefix Ausgang", "Dateiname", "Konvertierungsart"), show="headings")
self.target_files_treeview.heading("Prefix Eingang", text="Prefix Eingang")
self.target_files_treeview.heading("Prefix Ausgang", text="Prefix Ausgang")
self.target_files_treeview.heading("Dateiname", text="Dateiname")
self.target_files_treeview.heading("Konvertierungsart", text="Konvertierungsart")
self.target_files_treeview.column("Prefix Eingang", width=160, anchor="w")
self.target_files_treeview.column("Prefix Ausgang", width=160, anchor="w")
self.target_files_treeview.column("Dateiname", width=160, anchor="w")
self.target_files_treeview.column("Konvertierungsart", width=160, anchor="w")
self.target_files_treeview.place(x=10, y=170, width=650, height=380)
# Initialize default target file configurations (as in PowerShell script)
self.target_file_configurations = [
{"Prefix Eingang": "E", "Prefix Ausgang": "A", "Dateiname": "[Ursprungsdatei]_Simple", "Konvertierungsart": "Simpel"},
{"Prefix Eingang": "i", "Prefix Ausgang": "o", "Dateiname": "[Ursprungsdatei]_TIA", "Konvertierungsart": "TIA"},
{"Prefix Eingang": "K", "Prefix Ausgang": "M", "Dateiname": "[Ursprungsdatei]_WSCAD", "Konvertierungsart": "WSCAD"},
{"Prefix Eingang": "L", "Prefix Ausgang": "N", "Dateiname": "[Ursprungsdatei]_WSCAD_Bezug", "Konvertierungsart": "WSCAD mit Bezug"}
]
self._populate_target_files_treeview()
self.initialise_display()
def _populate_target_files_treeview(self):
# Clear existing data
for i in self.target_files_treeview.get_children():
self.target_files_treeview.delete(i)
# Populate with current configurations
for config in self.target_file_configurations:
self.target_files_treeview.insert("", "end", values=(
config.get("Prefix Eingang", ""),
config.get("Prefix Ausgang", ""),
config.get("Dateiname", ""),
config.get("Konvertierungsart", "")
))
def update_display(self, selection_type, path):
items = []
if os.path.exists(path) and os.path.isdir(path):
for item in os.listdir(path):
items.append(item)
elif not path: # Root drives
for part in psutil.disk_partitions():
if 'cdrom' not in part.opts and part.device:
items.append(part.mountpoint)
else: # Invalid path, go to drives
self.update_display(selection_type, "")
return
if selection_type == "txt":
self.txt_path_field.delete(0, tk.END)
self.txt_path_field.insert(0, path)
self.txt_list_box.delete(0, tk.END)
for item in sorted(items):
self.txt_list_box.insert(tk.END, item)
elif selection_type == "ver":
self.target_dir_path_field.delete(0, tk.END)
self.target_dir_path_field.insert(0, path)
self.target_dir_list_box.delete(0, tk.END)
for item in sorted(items):
self.target_dir_list_box.insert(tk.END, item)
def change_directory(self, selection_type, current_path, selected_item):
new_path = ""
if os.path.exists(current_path) and os.path.isdir(current_path):
new_path = os.path.join(current_path, selected_item)
elif not current_path: # Selected a drive
new_path = selected_item
else: # Invalid path, revert to drives
self.update_display(selection_type, "")
return
if os.path.isdir(new_path):
self.update_display(selection_type, new_path)
elif selection_type == "txt" and new_path.lower().endswith(".txt"):
self.txt_path_field.delete(0, tk.END)
self.txt_path_field.insert(0, new_path)
self.selected_txt_file = new_path # Store selected TXT file
self.switch_to_target_dir_tab()
else:
pass
def navigate_back(self, selection_type, current_path):
if len(current_path) > 3 and os.path.exists(current_path):
parent_dir = os.path.dirname(current_path)
self.update_display(selection_type, parent_dir)
elif len(current_path) <= 3 and current_path.endswith(":\\"):
self.update_display(selection_type, "")
def txt_path_field_return(self, event=None):
path = self.txt_path_field.get()
if os.path.isdir(path):
self.update_display("txt", path)
elif path.lower().endswith(".txt") and os.path.exists(path):
self.txt_path_field.delete(0, tk.END)
self.txt_path_field.insert(0, path)
self.selected_txt_file = path
self.switch_to_target_dir_tab()
else:
self.update_display("txt", "")
def txt_list_box_double_click(self, event=None):
selected_index = self.txt_list_box.curselection()
if selected_index:
selected_item = self.txt_list_box.get(selected_index[0])
self.change_directory("txt", self.txt_path_field.get(), selected_item)
def txt_list_box_back_key(self, event=None):
self.navigate_back("txt", self.txt_path_field.get())
def txt_back_button_click(self):
self.navigate_back("txt", self.txt_path_field.get())
def txt_select_button_click(self):
selected_index = self.txt_list_box.curselection()
if selected_index:
selected_item = self.txt_list_box.get(selected_index[0])
current_path = self.txt_path_field.get()
potential_file_path = os.path.join(current_path, selected_item)
if potential_file_path.lower().endswith(".txt") and os.path.exists(potential_file_path):
self.txt_path_field.delete(0, tk.END)
self.txt_path_field.insert(0, potential_file_path)
self.selected_txt_file = potential_file_path
self.switch_to_target_dir_tab()
else:
self.change_directory("txt", current_path, selected_item)
def target_dir_path_field_return(self, event=None):
path = self.target_dir_path_field.get()
if os.path.isdir(path):
self.update_display("ver", path)
else:
self.update_display("ver", "")
def target_dir_list_box_double_click(self, event=None):
selected_index = self.target_dir_list_box.curselection()
if selected_index:
selected_item = self.target_dir_list_box.get(selected_index[0])
self.change_directory("ver", self.target_dir_path_field.get(), selected_item)
def target_dir_list_box_back_key(self, event=None):
self.navigate_back("ver", self.target_dir_path_field.get())
def target_dir_back_button_click(self):
self.navigate_back("ver", self.target_dir_path_field.get())
def target_dir_select_button_click(self):
selected_index = self.target_dir_list_box.curselection()
if selected_index:
selected_item = self.target_dir_list_box.get(selected_index[0])
current_path = self.target_dir_path_field.get()
potential_dir_path = os.path.join(current_path, selected_item)
if os.path.isdir(potential_dir_path):
self.change_directory("ver", current_path, selected_item)
else:
self.confirm_target_directory(current_path, selected_item)
def target_dir_confirm_button_click(self):
selected_index = self.target_dir_list_box.curselection()
selected_item = ""
if selected_index:
selected_item = self.target_dir_list_box.get(selected_index[0])
current_path = self.target_dir_path_field.get()
self.confirm_target_directory(current_path, selected_item)
def confirm_target_directory(self, path, selected_item):
full_path = os.path.join(path, selected_item) if selected_item else path
if os.path.isdir(full_path):
self.target_dir_path_field.delete(0, tk.END)
self.target_dir_path_field.insert(0, full_path)
self.selected_target_directory = full_path # Store selected target directory
self.switch_to_progress_tab()
self.start_excel_conversion() # Start conversion here
elif not full_path and not selected_item:
if os.path.isdir(path):
self.selected_target_directory = path
self.switch_to_progress_tab()
self.start_excel_conversion() # Start conversion here
else:
self.update_display("ver", "")
else:
self.update_display("ver", "")
def switch_to_target_dir_tab(self):
self.notebook.select(self.tab_target_dir_select)
# Determine initial path for target directory selection based on settings
target_path_option = self.target_path_option_var.get()
initial_target_path = ""
if target_path_option == "WieTXT":
if self.selected_txt_file:
initial_target_path = os.path.dirname(self.selected_txt_file)
else:
initial_target_path = os.getcwd()
elif target_path_option == "WieZielDatei":
# This option is a bit ambiguous in the PowerShell context without a previous "ZielDatei"
# For simplicity, if a TXT file is selected, use its directory. Otherwise, current working directory.
if self.selected_txt_file:
initial_target_path = os.path.dirname(self.selected_txt_file)
else:
initial_target_path = os.getcwd()
elif target_path_option == "EigenerPfad":
initial_target_path = self.target_dir_start_path_label.cget("text")
if not initial_target_path or not os.path.isdir(initial_target_path):
initial_target_path = os.getcwd() # Fallback
self.update_display("ver", initial_target_path)
self.current_task_label.config(text="Zielverzeichnis auswaehlen")
def switch_to_progress_tab(self):
self.notebook.select(self.tab_progress)
self.current_task_label.config(text="Excel-Dateien erstellen")
self.completed_tasks_listbox.delete(0, tk.END) # Clear previous tasks
self.progress_bar["value"] = 0
self.processed_entries_label.config(text="0")
self.total_entries_label.config(text="0")
def start_excel_conversion(self):
if not self.selected_txt_file or not os.path.exists(self.selected_txt_file):
print("No TXT file selected or file does not exist.")
self.completed_tasks_listbox.insert(tk.END, "FEHLER: Keine TXT-Datei ausgewaehlt oder Datei existiert nicht.")
return
if not self.selected_target_directory or not os.path.isdir(self.selected_target_directory):
print("No target directory selected or directory does not exist.")
self.completed_tasks_listbox.insert(tk.END, "FEHLER: Kein Zielverzeichnis ausgewaehlt oder Verzeichnis existiert nicht.")
return
all_duplicate_inputs = []
all_duplicate_outputs = []
all_errors = []
# Read total entries from the TXT file for progress bar calculation
try:
with open(self.selected_txt_file, 'r', encoding='utf-8') as f:
total_lines = len(f.readlines())
self.total_entries_label.config(text=str(total_lines))
except Exception as e:
print(f"Error reading TXT file for total entries: {e}")
total_lines = 1 # Avoid division by zero
self.progress_bar["maximum"] = len(self.target_file_configurations)
processed_files_count = 0
for config in self.target_file_configurations:
prefix_eingang = config.get("Prefix Eingang", "")
prefix_ausgang = config.get("Prefix Ausgang", "")
dateiname_template = config.get("Dateiname", "")
konvertierungsart = config.get("Konvertierungsart", "")
ziel_datei = self.excel_converter._excel_datei_name_erstellen(
dateiname_template,
self.selected_txt_file,
self.selected_target_directory
)
self.current_task_label.config(text=f"Erstelle Excel-Datei: {os.path.basename(ziel_datei)}")
self.master.update_idletasks() # Update GUI
try:
gleiche_eingaenge = []
gleiche_ausgaenge = []
fehler = []
if konvertierungsart == "Simpel":
gleiche_eingaenge, gleiche_ausgaenge, fehler = self.excel_converter.prep_simpel(
self.selected_txt_file, prefix_eingang, prefix_ausgang, ziel_datei
)
elif konvertierungsart == "TIA":
gleiche_eingaenge, gleiche_ausgaenge, fehler = self.excel_converter.prep_tia(
self.selected_txt_file, prefix_eingang, prefix_ausgang, ziel_datei
)
elif konvertierungsart == "WSCAD":
gleiche_eingaenge, gleiche_ausgaenge, fehler = self.excel_converter.prep_wscad(
self.selected_txt_file, prefix_eingang, prefix_ausgang, ziel_datei
)
elif konvertierungsart == "WSCAD mit Bezug":
gleiche_eingaenge, gleiche_ausgaenge, fehler = self.excel_converter.prep_wscad_bezug(
self.selected_txt_file, prefix_eingang, prefix_ausgang, ziel_datei
)
all_duplicate_inputs.extend(gleiche_eingaenge)
all_duplicate_outputs.extend(gleiche_ausgaenge)
all_errors.extend(fehler)
self.completed_tasks_listbox.insert(tk.END, f"ERFOLG: {os.path.basename(ziel_datei)}")
except Exception as e:
self.completed_tasks_listbox.insert(tk.END, f"FEHLER bei {os.path.basename(ziel_datei)}: {e}")
print(f"Error converting {os.path.basename(ziel_datei)}: {e}")
processed_files_count += 1
self.progress_bar["value"] = processed_files_count
self.processed_entries_label.config(text=str(total_lines)) # For now, set to total lines after each file for simplicity
self.master.update_idletasks() # Update GUI
self.switch_to_duplicates_tab(all_duplicate_inputs, all_duplicate_outputs, all_errors)
def switch_to_duplicates_tab(self, duplicate_inputs, duplicate_outputs, errors):
self.notebook.select(self.tab_duplicates)
self.current_task_label.config(text="Ausgabe Duplikate und Fehler")
# Clear existing data
for i in self.duplicate_inputs_treeview.get_children():
self.duplicate_inputs_treeview.delete(i)
for i in self.duplicate_outputs_treeview.get_children():
self.duplicate_outputs_treeview.delete(i)
for i in self.unassigned_errors_treeview.get_children():
self.unassigned_errors_treeview.delete(i)
# Populate with new data
for item in duplicate_inputs:
# Ensure 'Name' and 'Zeile' keys exist, provide default if not
name = item.get('Name', '')
zeile = item.get('Zeile', '')
self.duplicate_inputs_treeview.insert("", "end", values=(zeile, name))
for item in duplicate_outputs:
name = item.get('Name', '')
zeile = item.get('Zeile', '')
self.duplicate_outputs_treeview.insert("", "end", values=(zeile, name))
for item in errors:
name = item.get('Name', '')
zeile = item.get('Zeile', '')
self.unassigned_errors_treeview.insert("", "end", values=(zeile, name))
def initialise_display(self):
self.update_display("txt", os.getcwd())
def change_txt_start_path(self):
new_path = filedialog.askdirectory()
if new_path:
self.txt_start_path_label.config(text=new_path)
def save_radio_button_settings(self):
selected_option = self.target_path_option_var.get()
print(f"Selected target path option: {selected_option}")
def change_target_dir_start_path(self):
new_path = filedialog.askdirectory()
if new_path:
self.target_dir_start_path_label.config(text=new_path)
def save_target_file_settings(self):
# Read current data from Treeview
updated_configurations = []
for item_id in self.target_files_treeview.get_children():
values = self.target_files_treeview.item(item_id, 'values')
if len(values) == 4:
updated_configurations.append({
"Prefix Eingang": values[0],
"Prefix Ausgang": values[1],
"Dateiname": values[2],
"Konvertierungsart": values[3]
})
self.target_file_configurations = updated_configurations
print("Target file settings saved.")
print(self.target_file_configurations)
def call_portal_to_excel(self, textlines, konvertierungsart):
ec = ExcelConverter()
all_duplicate_inputs = []
all_duplicate_outputs = []
all_errors = []
prefix_eingang = "E"
prefix_ausgang = "A"
target_path = os.environ.get("PROJECT_WORK")
#self.current_task_label.config(text=f"Erstelle Excel-Datei: {os.path.basename(ziel_datei)}")
gleiche_eingaenge = []
gleiche_ausgaenge = []
fehler = []
if konvertierungsart == "Simpel":
# df_eingaenge, df_ausgaenge wird in excel Datei geschrieben. Hier zu Ausgabe hinzugefügt, damit unittest befüllt werden kann
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge = ec._prep_simpel(
textlines, prefix_eingang, prefix_ausgang
)
return df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge
elif konvertierungsart == "TIA":
# df_consts und df_tags wird in excel Datei geschrieben. Hier zu Ausgabe hinzugefügt, damit unittest befüllt werden kann
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_consts, df_tags = ec._prep_tia(
textlines, prefix_eingang, prefix_ausgang
)
return(df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_consts, df_tags)
elif konvertierungsart == "WSCAD" or "WSCAD mit Bezug":
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge = ec._prep_wscad(
textlines, prefix_eingang, prefix_ausgang
)
return(df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge)
all_duplicate_inputs.extend(gleiche_eingaenge)
all_duplicate_outputs.extend(gleiche_ausgaenge)
all_errors.extend(fehler)
#self.completed_tasks_listbox.insert(tk.END, f"ERFOLG: {os.path.basename(ziel_datei)}")
#self.switch_to_duplicates_tab(all_duplicate_inputs, all_duplicate_outputs, all_errors)
class TestPlant(unittest.TestCase):
def test_Simpel(self):
textlines = """'DI0023','STEUERSPANNUNG 24V UC200 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC200 OK','CONTROL VOLTAGE 24V UC200 OK','TENSION DE MANIOBRA 24V UC200 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'XY0022','STEUERSPANNUNG 24V UC103 ','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC103 OK','CONTROL VOLTAGE 24V UC103 OK','TENSION DE MANIOBRA 24V UC103 OK',''
'DI0021','STEUERSPANNUNG 24V UC102-UC101 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC102-UC101 OK','CONTROL VOLTAGE 24V UC102-UC101 OK','TENSION DE MANIOBRA 24V UC102-UC101 OK',''
'KF0013','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','COMPR. AIR OFF','=A01+UH000-KF01Q','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','CLOSE PRESS.-MAINTENANCE UNIT','AIRE COMPRIMIDO/VÁLVULA DE AISLAMIENTO C',''
'KF0012','RESET NOT HALT','RESET E-STOP','=A01+UH000-KF01Q','RESET NOT HALT','RESET EMERGENCY STOP','RESET PARO DE EMERGENCIA',''
'PF0011','ÜBERSTROM/SICHERUNGSFALL','IL-YE-LED','=A01+UH000-KF01Q','ÜBERSTROM/SICHERUNGSFALL','OVERCURRENT/MIN. CIRCUIT BREAKERS','SOBREINTENSIDAD/FUSIBLE AUTOMÁTICO',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
"""
textarray = textlines.splitlines()
'''Testcase: 10 Zeilen mit I/O: Davon 4 Eingänge (1 Duplikat), 5 Ausgänge (1 Duplikat), 1 Fehlerhafte BMK'''
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge = call_portal_to_excel(self, textarray, "Simpel")
# print("\n Eingänge:\n")
# print(df_eingaenge)
# print("\n Ausgänge:\n")
# print(df_ausgaenge)
# print("\n gleiche Eingänge:\n")
# print(df_gleiche_eingaenge)
# print("\n gleiche Ausgänge:\n")
# print(df_gleiche_ausgaenge)
self.assertEqual(len(df_fehler), 1) # Fehler anzeigen: Zeile 4 Prefix XY
self.assertEqual(len(df_eingaenge), 4 - len(df_gleiche_eingaenge)) # 4 Eingänge in Liste, einer davon als Duplikat
self.assertEqual(len(df_ausgaenge), 5 - len(df_gleiche_ausgaenge)) # 5 Ausgänge in Liste, einer davon Duplikat
def test_TIA(self):
textlines = """'DI0023','STEUERSPANNUNG 24V UC200 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC200 OK','CONTROL VOLTAGE 24V UC200 OK','TENSION DE MANIOBRA 24V UC200 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'XY0022','STEUERSPANNUNG 24V UC103 ','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC103 OK','CONTROL VOLTAGE 24V UC103 OK','TENSION DE MANIOBRA 24V UC103 OK',''
'DI0021','STEUERSPANNUNG 24V UC102-UC101 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC102-UC101 OK','CONTROL VOLTAGE 24V UC102-UC101 OK','TENSION DE MANIOBRA 24V UC102-UC101 OK',''
'KF0013','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','COMPR. AIR OFF','=A01+UH000-KF01Q','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','CLOSE PRESS.-MAINTENANCE UNIT','AIRE COMPRIMIDO/VÁLVULA DE AISLAMIENTO C',''
'KF0012','RESET NOT HALT','RESET E-STOP','=A01+UH000-KF01Q','RESET NOT HALT','RESET EMERGENCY STOP','RESET PARO DE EMERGENCIA',''
'PF0011','ÜBERSTROM/SICHERUNGSFALL','IL-YE-LED','=A01+UH000-KF01Q','ÜBERSTROM/SICHERUNGSFALL','OVERCURRENT/MIN. CIRCUIT BREAKERS','SOBREINTENSIDAD/FUSIBLE AUTOMÁTICO',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
"""
textarray = textlines.splitlines()
'''Testcase: 10 Zeilen mit I/O: Davon 4 Eingänge (1 Duplikat), 5 Ausgänge (1 Duplikat), 1 Fehlerhafte BMK'''
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_consts, df_tags = call_portal_to_excel(self, textarray, "TIA")
# print("\n Consts:\n")
# print(df_consts)
# print("\n Tags:\n")
# print(df_tags)
# print("\n gleiche Eingänge:\n")
# print(df_gleiche_eingaenge)
# print("\n gleiche Ausgänge:\n")
# print(df_gleiche_ausgaenge)
self.assertIn("DI0020", df_gleiche_eingaenge["NameOriginal"].values) # Püfen, ob doppelter Name in doppelteEingänge auftaucht
self.assertIn("MA1020", df_gleiche_ausgaenge["NameOriginal"].values) # Püfen, ob doppelter Name in doppelteAusgänge auftaucht
self.assertEqual(len(df_fehler), 1) # Fehler anzeigen: Zeile 4 Prefix XY
# Test, ob länge passt. Eigentlich 10 Eingabezeilen, Fehler bleibt drin, aber 2x 1 Duplikat in Eingang und Ausgang abziehen
self.assertEqual(len(df_consts), len(textarray)-(len(df_gleiche_ausgaenge)+len(df_gleiche_eingaenge)))
self.assertEqual(len(df_tags), len(textarray)-(len(df_gleiche_ausgaenge)+len(df_gleiche_eingaenge)))
def test_WSCAD(self):
textlines = """'DI0023','STEUERSPANNUNG 24V UC200 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC200 OK','CONTROL VOLTAGE 24V UC200 OK','TENSION DE MANIOBRA 24V UC200 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'XY0022','STEUERSPANNUNG 24V UC103 ','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC103 OK','CONTROL VOLTAGE 24V UC103 OK','TENSION DE MANIOBRA 24V UC103 OK',''
'DI0021','STEUERSPANNUNG 24V UC102-UC101 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC102-UC101 OK','CONTROL VOLTAGE 24V UC102-UC101 OK','TENSION DE MANIOBRA 24V UC102-UC101 OK',''
'KF0013','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','COMPR. AIR OFF','=A01+UH000-KF01Q','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','CLOSE PRESS.-MAINTENANCE UNIT','AIRE COMPRIMIDO/VÁLVULA DE AISLAMIENTO C',''
'KF0012','RESET NOT HALT','RESET E-STOP','=A01+UH000-KF01Q','RESET NOT HALT','RESET EMERGENCY STOP','RESET PARO DE EMERGENCIA',''
'PF0011','ÜBERSTROM/SICHERUNGSFALL','IL-YE-LED','=A01+UH000-KF01Q','ÜBERSTROM/SICHERUNGSFALL','OVERCURRENT/MIN. CIRCUIT BREAKERS','SOBREINTENSIDAD/FUSIBLE AUTOMÁTICO',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
"""
textarray = textlines.splitlines()
'''Testcase: 10 Zeilen mit I/O: Davon 4 Eingänge (1 Duplikat), 5 Ausgänge (1 Duplikat), 1 Fehlerhafte BMK'''
df_gleiche_eingaenge, df_gleiche_ausgaenge, df_fehler, df_eingaenge, df_ausgaenge = call_portal_to_excel(self, textarray, "WSCAD mit Bezug")
print("\n Eingänge:\n")
print(df_eingaenge)
print("\n Ausgänge:\n")
print(df_ausgaenge)
# print("\n Fehler")
# print(df_fehler)
# print("\n gleiche Eingänge:\n")
# print(df_gleiche_eingaenge)
# print("\n gleiche Ausgänge:\n")
# print(df_gleiche_ausgaenge)
if __name__ == "__main__":
suite = unittest.TestSuite()
suite.addTest(TestPlant('test_Simpel'))
suite.addTest(TestPlant('test_TIA'))
suite.addTest(TestPlant('test_WSCAD'))
runner = unittest.TextTestRunner()
runner.run(suite)
export_type_arr = ("Simpel", "TIA", "WSCAD", "WSCAD mit Bezug")
textlines = """'DI0023','STEUERSPANNUNG 24V UC200 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC200 OK','CONTROL VOLTAGE 24V UC200 OK','TENSION DE MANIOBRA 24V UC200 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'DI0020','STEUERSPANNUNG 24V UH000 OK','24V OK UH000','=A01+UH000-KF02i','STEUERSPANNUNG 24V UH000 OK','CONTROL VOLTAGE 24V UH000 OK','TENSION DE MANIOBRA 24V UH000 OK',''
'XY0022','STEUERSPANNUNG 24V UC103 ','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC103 OK','CONTROL VOLTAGE 24V UC103 OK','TENSION DE MANIOBRA 24V UC103 OK',''
'DI0021','STEUERSPANNUNG 24V UC102-UC101 OK','24V OK','=A01+UH000-KF02i','STEUERSPANNUNG 24V UC102-UC101 OK','CONTROL VOLTAGE 24V UC102-UC101 OK','TENSION DE MANIOBRA 24V UC102-UC101 OK',''
'KF0013','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','COMPR. AIR OFF','=A01+UH000-KF01Q','DRUCKL.-WARTUNGSEINH. SCHLIESSEN','CLOSE PRESS.-MAINTENANCE UNIT','AIRE COMPRIMIDO/VÁLVULA DE AISLAMIENTO C',''
'KF0012','RESET NOT HALT','RESET E-STOP','=A01+UH000-KF01Q','RESET NOT HALT','RESET EMERGENCY STOP','RESET PARO DE EMERGENCIA',''
'PF0011','ÜBERSTROM/SICHERUNGSFALL','IL-YE-LED','=A01+UH000-KF01Q','ÜBERSTROM/SICHERUNGSFALL','OVERCURRENT/MIN. CIRCUIT BREAKERS','SOBREINTENSIDAD/FUSIBLE AUTOMÁTICO',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
'MA1020','ANTRIEB PUFFER F0101 FLYER 1','M-BI-LE','=A01+UC101-KF02Q','','','',''
"""
textarray = textlines.splitlines()
exporter = ExcelConverter()
for etype in export_type_arr:
export_path = f"work/excel_{etype}.xlsx"
exporter.export_excel(textarray, etype, export_path)
Reference in New Issue View Git Blame Copy Permalink