s.steuer/SchnittstellenanpassungRDSivas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2668eb89f80d4dbb15fc87c773d129e415f506ee
SchnittstellenanpassungRDSivas/lib/process_bom.py
T

1009 lines
50 KiB
Python
Raw Blame History

import argparse
import csv
import os
import sqlite3
def main():
parser = argparse.ArgumentParser(description='Programm erstellt finale SIVAS-Stückliste aus der RuleDesigner-Stücklisten-CSV unter Berücksichtigung der spezifizierten Verschiebelogik!',
prog='process_bom.py',
formatter_class=lambda prog: argparse.HelpFormatter(prog, max_help_position=50))
parser.register("action", "store_after_check", ArgparseDirectoryCheck)
parser.register("action", "store_after_code_check", ArgparseParentcodeCheck)
parser.add_argument("parent_code", action="store_after_code_check", type=str, help="Teilenummer der Baugruppe")
parser.add_argument("-csv_dir", action="store_after_check", default=os.environ.get("PROJECT_DATA_IN"), metavar="dir", help="Verzeichnis, aus dem die Stücklisten-CSV aus RuleDesigner eingelesen werden soll. Default: /data")
parser.add_argument("-bom", action="store_true", help="Ausgabe der neuen, aufbereiteten Stückliste")
parser.add_argument("-av", action="store_true", help="Ausgabe der AV-Workflowmeldungen (falls vorhanden)")
parser.add_argument("-tos", action="store_true", help="Ausgabe der TOS-Workflowmeldungen (falls vorhanden)")
subparsers = parser.add_subparsers(title="output type", required=True, help="Legt fest, in welcher Form die Stückliste ausgegeben werden soll. cmd = In Konsole, csv = als csv-Datei", dest="output_type")
sp_cmd = subparsers.add_parser("cmd", help="Ausgabe der über die Flags -bom / -av / -tos ausgewählten Elemente in der Konsole")
sp_csv = subparsers.add_parser("csv", help="Ausgabe der über die Flags -bom / -av / -tos ausgewählten Elemente als CSV-Datei(en)")
sp_csv.register("action", "store_after_check", ArgparseDirectoryCheck)
sp_csv.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben")
sp_cmd.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben")
sp_csv.add_argument("-od", "--out_dir", action="store_after_check", default=os.environ.get("PROJECT_DATA_OUT"), metavar="dir", help="Ausgabeverzeichnis fertigen Stücklisten-CSV")
args: dict = vars(parser.parse_args())
if os.environ.get("DB_PATH") is not None:
# nur 1 Datenbankinstanz über gesamte Laufzeit (ähnlich Singleton)
db = Database(os.environ.get("DB_PATH"))
else:
raise ValueError("Bitte bin/setenv.bat vor diesem Programm für erforderliche Umgebungsvariablen ausführen!")
try:
old_bom = COMP(db)
new_bom = RD_COMP(db, args["parent_code"])
csv_handler = CSVHandler()
sql_handler = SQLFileHandler()
bom_processing = BOMProcessing(db, args["parent_code"])
av_workflow_msg = AVWorkflowMessage(db, args["parent_code"])
tos_workflow_msg = TOSWorkflowMessage(db, args["parent_code"])
# SIVAS "nachbauen"
sql_source_filename = "sivas_replika.sql"
sivas_source_data = sql_handler.read_from_file(filename=sql_source_filename)
old_bom.rebuild_sivas(sivas_source_data)
# CSV-Daten in Datenbank schreiben
csv_data = csv_handler.read_as_dict(filename=f"BOM_{args["parent_code"]}")
new_bom.create_tmp_table()
new_bom.insert_csv_data(csv_data)
# Positionen für AV- bzw. TOS-Workflowmeldungen generieren (muss vor dem Schreiben der neuen Stücklist erfolgen!)
av_workflow_msg.get_av_workflow_posnrs_and_codes()
tos_workflow_msg.get_tos_workflow_posnrs_and_codes()
# Neue Stückliste erstellen / schreiben
bom_processing.write_new_bom()
# Ausgabe in Konsole
if args["output_type"] == "cmd":
if args["bom"]:
if args["verbose"]:
print(repr(new_bom))
else:
print(new_bom)
if args["av"]:
print(av_workflow_msg)
if args["tos"]:
print(tos_workflow_msg)
# Ausgabe als CSV-Datei(en)
elif args["output_type"] == "csv":
filedir = args["out_dir"]
filenames = []
file_data = []
# CSV-Ausgabe der aufbereiteten Stückliste
if args["bom"]:
if args["verbose"]:
bom_data = new_bom.get_verbose_data()
else:
bom_data = new_bom.get_compact_data()
filenames.append(f"BOM_{args["parent_code"]}_new.csv")
file_data.append(bom_data)
# CSV-Ausgabe der Positionsnummern aus der alten Stückliste, die Bestandteil der AV-Workflowmeldung sind
if args["av"]:
msg_data = av_workflow_msg.get_data()
filenames.append(f"AV_{args["parent_code"]}.csv")
file_data.append(msg_data)
# CSV-Ausgabe der Positionsnummern aus der alten Stückliste, die Bestandteil der TOS-Workflowmeldung sind
if args["tos"]:
msg_data = tos_workflow_msg.get_data()
filenames.append(f"TOS_{args["parent_code"]}.csv")
file_data.append(msg_data)
for name, data in zip(filenames, file_data):
csv_handler.write_from_dict(data=data, filedir=filedir, filename=name)
else:
NotImplementedError
finally:
db.close()
db.remove() # Damit bei jedem Lauf eine neue Datenbank angelegt wird -> mit :memory: traten Bugs auf
class ArgparseDirectoryCheck(argparse.Action):
def __call__(self, parser, namespace, values: list[str], option_string = None):
if not os.path.isdir(values[0]):
raise NotADirectoryError
setattr(namespace, self.dest, values[0])
class ArgparseParentcodeCheck(argparse.Action):
def __call__(self, parser, namespace, values: str, option_string = None):
csv_filepath = os.path.join(os.environ.get("PROJECT_DATA_IN"), f"BOM_{values}.csv")
if not os.path.isfile(csv_filepath):
raise FileNotFoundError(csv_filepath)
setattr(namespace, self.dest, values)
class FileHandler:
def check_filedir(self, filedir: str) -> str:
if filedir is None:
raise ValueError("Bitte bin/setenv.bat ausführen, oder Verzeichnis angegeben, in dem die Eingabedatei liegt!")
if not os.path.isdir(filedir):
raise NotADirectoryError
return filedir
def check_filepath(self, filepath: str) -> str:
if not os.path.isfile(filepath):
raise FileNotFoundError("Angegebener Dateiname existiert nicht!")
return filepath
class CSVHandler(FileHandler):
def __init__(self, encoding="utf-8", delimiter: str = ";"):
self.delimiter = delimiter
self.encoding = encoding
def read_as_dict(self, filename: str, filedir: str = os.environ.get("PROJECT_DATA_IN"), ) -> list[dict[str]]:
filename = f"{filename.removesuffix(".csv")}.csv"
filedir = self.check_filedir(filedir)
filepath = os.path.join(filedir, filename)
if not filepath:
raise FileNotFoundError("Zur angegebenen Teilenummer existiert keine Stücklisten-CSV!")
with open(filepath, "r", encoding=self.encoding, newline="") as csv_file:
reader = csv.DictReader(csv_file, delimiter=self.delimiter)
required_col_names = ["ParentCode", "Positionsnummer", "ChildCode", "Menge", "Uebergabelaenge"]
if reader.fieldnames != required_col_names:
raise ValueError(f"Spaltennamen falsch. Korrekt: {required_col_names}")
csv_data = []
for line in reader:
line["Positionsnummer"] += "0" # Angleichung Positionsnummer an SIVAS-Format (Pos. * 10)
csv_data.append(line)
return csv_data
def write_from_dict(self, filename: str, data: list[dict[str]], filedir: str = os.environ.get("PROJECT_DATA_OUT")) -> None:
filedir = self.check_filedir(filedir)
filepath = os.path.join(filedir, f"{filename.removesuffix(".csv")}.csv")
with open(filepath, "w", encoding=self.encoding, newline="") as csv_file:
writer = csv.DictWriter(csv_file,
fieldnames=data[0].keys(),
delimiter=self.delimiter)
writer.writeheader()
writer.writerows(data)
self.check_filepath(filepath)
print(f"CSV geschrieben: {filepath}")
class SQLFileHandler(FileHandler):
def __init__(self, encoding: str = "utf-8"):
self.encoding = encoding
def read_from_file(self, filename: str, filedir: str = os.environ.get("PROJECT_DATA_IN")) -> str:
filedir = self.check_filedir(filedir)
filepath = os.path.join(filedir, filename)
filepath = self.check_filepath(filepath)
with open(filepath, "r", encoding="utf-8", newline="") as sql_file:
sql = sql_file.read()
return sql
class Database:
def __init__(self, db_path: str):
if not os.path.isdir(os.path.dirname(db_path)):
raise NotADirectoryError
self.db_path = db_path
self.con = sqlite3.connect(self.db_path, autocommit=True)
self.con.row_factory = self.dict_factory
@staticmethod
def dict_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
d[col[0]] = row[idx]
return d
def close(self):
self.con.close()
def remove(self):
if os.path.isfile(self.db_path):
os.remove(self.db_path)
class RD_COMP:
"""
Temporäre Tabelle, in die die Daten der Stücklisten-CSV aus RuleDesigner geschrieben werden
Tabelle wird für die Verarbeitung der Stückliste benötigt. Sobald die finale Stückliste steht und in die Tabelle 'COMP'
wird sie nicht mehr benötigt.
"""
def __init__(self, db: Database, parent_code: str):
self.db = db
self.parent_code = parent_code
def __repr__(self):
res = self.get_verbose_data()
output = self._sql_as_ascii_table(res)
return output
def __str__(self):
res = self.get_compact_data()
output = self._sql_as_ascii_table(res)
return output
def create_tmp_table(self) -> None:
"""Schreibt neue Stücklistendaten aus CSV in temporäre Datenbanktabelle 'RD_COMP'."""
self.db.con.execute("""CREATE TABLE RD_COMP (ParentCode TEXT NOT NULL,
Positionsnummer INTEGER NOT NULL,
ChildCode TEXT NOT NULL,
Menge INTEGER NOT NULL,
Uebergabelaenge REAL NOT NULL,
PRIMARY KEY(Positionsnummer)
)
""")
def drop(self):
self.db.con.execute("DROP TABLE RD_COMP")
def insert_csv_data(self, csv_data: list[dict[str]]):
self.db.con.executemany("""
INSERT INTO RD_COMP (ParentCode, Positionsnummer, ChildCode, Menge, Uebergabelaenge)
VALUES (:ParentCode, :Positionsnummer, :ChildCode, :Menge, :Uebergabelaenge)
""", csv_data)
def get_verbose_data(self) -> list[dict[str]]:
res = self.db.con.execute("""
SELECT C.*, R.LFD_NR, B.UEB_LEISTEN_NR
FROM COMP C
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON C.UEBTEILENR = B.UEBTEILENR
AND C.POSNR = B.POSNR
LEFT JOIN STKL_REGEL R
ON C.UEBTEILENR = R.UEBTEILENR
AND C.POSNR = R.POSNR
WHERE C.UEBTEILENR = :parent_code
ORDER BY C.POSNR
""", {"parent_code": self.parent_code}).fetchall()
return res
def get_compact_data(self) -> list[dict[str]]:
res = self.db.con.execute("""
SELECT C.POSNR, C.KOMPTEILENR, C.ABW_ANZAHL, C.MENGE, C.LAENGE, C.ZUSCHNITT, R.LFD_NR, B.UEB_LEISTEN_NR
FROM COMP C
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON C.UEBTEILENR = B.UEBTEILENR
AND C.POSNR = B.POSNR
LEFT JOIN STKL_REGEL R
ON C.UEBTEILENR = R.UEBTEILENR
AND C.POSNR = R.POSNR
WHERE C.UEBTEILENR = :parent_code
ORDER BY C.POSNR
""", {"parent_code": self.parent_code}).fetchall()
return res
def _sql_as_ascii_table(self, sql_data: list[dict[str]]):
"""Daten aus NEW-BOM als Tabelle ausgeben"""
headers = list(sql_data[0].keys())
rows = [[str(row.get(h, "") if row.get(h, "") is not None else "") for h in headers] for row in sql_data]
col_widths = [max(len(str(h)), max(len(str(row[i])) for row in rows)) for i, h in enumerate(headers)]
header_line = " | ".join(f"{h:<{col_widths[i]}}" for i, h in enumerate(headers))
separator = "-+-".join("-" * col_widths[i] for i in range(len(headers)))
row_lines = [" | ".join(f"{row[i]:<{col_widths[i]}}" for i in range(len(headers))) for row in rows]
return "\n".join([header_line, separator] + row_lines)
class COMP:
def __init__(self, db: Database):
self.db = db
def rebuild_sivas(self, sql: str) -> None:
self.db.con.executescript(sql)
class BOMProcessing:
def __init__(self, db: Database, parent_code: str):
self.db = db
self.parent_code = parent_code
def _get_codes_case_12_or_13(self) -> None:
"""
Ermittelt alle Teilenummern, die unter Fall 1.2 oder Fall 1.3 fallen.
Es wird dafür geprüft, welche Teilenummern NICHT unter Fall 1.1 fallen.
Wieso wird nicht geprüft, welche Teilenummern in Fall 1.1 liegen?
- Weil die SQL-Abfrage dann komplexer wird.
Die Prüfung läuft so ab:
1. Teilenr-Längen-Kombinationen aus der alten und neuen Stückliste mit UNION ALL verschmelzen
2. Mit COUNT ermitteln, wie oft jede Teilenummer-Längen-Kombination in der verschmolzenen Stückliste auftritt
3. Die Teilenummern in die temporäre Tabelle TMP_CODES_CASE_12_OR_13 schreiben, sobald 1x eine Teilenr-Längen-Kombination auftritt, die nur 1x vorkommt
=> Diese Teilenummern unterliegen entweder Fall 1.2 oder Fall 1.3, aber definitiv nicht Fall 1.1, weil eine Zuordnung zu Fall 1.1 nur dann möglich ist,
wenn alle Teilenr-Längen-Kombinationen einer Teilenummer exakt 2x auftreten!
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_CODES_CASE_12_OR_13")
self.db.con.execute("CREATE TABLE TMP_CODES_CASE_12_OR_13 (KOMPTEILENR TEXT NOT NULL)")
self.db.con.execute("""
WITH combined_bom AS (
SELECT C.KOMPTEILENR, COALESCE(C.LAENGE, 0) AS LAENGE
FROM COMP C
WHERE C.UEBTEILENR = :parent_code
UNION ALL
SELECT ChildCode, Uebergabelaenge
FROM RD_COMP
WHERE ParentCode = :parent_code
)
INSERT INTO TMP_CODES_CASE_12_OR_13
SELECT DISTINCT KOMPTEILENR
FROM (
SELECT KOMPTEILENR, LAENGE, COUNT(*) AS ANZAHL
FROM combined_bom
GROUP BY KOMPTEILENR, LAENGE
)
WHERE ANZAHL = 1
""", {"parent_code": self.parent_code})
def _get_codes_case_13(self) -> None:
"""
Ermittelt alle Teilenummern, die unter Fall 1.3 fallen
Fall 1.3 umfasst alle Positionen, die entweder in der neuen Stückliste dazugekommen sind (Fall 1.31) oder deren
Teilenr-Längen-Kombination in beiden Stücklisten vorliegen (Fall 1.32).
Wenn Positionen bereits unter Fall 1.1 oder Fall 1.2 fallen, dann werden sie NICHT Fall 1.32 zugeordnet, selbst wenn
Fall 1.32 auf diese Positionen zutreffen würde.
Die Prüfung läuft so ab:
1. Alle Teilenr-Posnr-Kombinationen aus der alten und neuen Stückliste mit UNION ALL verschmelzen, sofern sich die Teilenummern in der
Liste an Teilenummern befindet, die unter Fall 1.2 oder Fall 1.3 fallen (siehe Methode _get_codes_case_12_or_13())
2. Mit COUNT ermitteln, wie oft jede Teilenr-Posnr-Kombination in der zusammengefügten Stückliste auftritt
3. Die Teilenummern, deren Teilenr-Posnr-Kombination nur 1x auftritt, in die temporäre Tabelle TMP_CODES_CASE_13 schreiben
=> Diese Teilenummern unterliegen definitv Fall 1.3, weil in Fall 1.2 alle Teilenr-Posnr-Kombinationen einer Teilenummer exakt 2x auftreten müssen!
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_CODES_CASE_13")
self.db.con.execute("CREATE TABLE TMP_CODES_CASE_13 (KOMPTEILENR TEXT NOT NULL)")
self.db.con.execute("""
WITH combined_bom AS (
SELECT KOMPTEILENR, POSNR
FROM COMP
WHERE UEBTEILENR = :parent_code
AND KOMPTEILENR IN (SELECT * FROM TMP_CODES_CASE_12_OR_13)
UNION ALL
SELECT ChildCode, Positionsnummer
FROM RD_COMP
WHERE ParentCode = :parent_code
AND ChildCode IN (SELECT * FROM TMP_CODES_CASE_12_OR_13)
)
INSERT INTO TMP_CODES_CASE_13 (KOMPTEILENR)
SELECT DISTINCT KOMPTEILENR
FROM (
SELECT KOMPTEILENR, COUNT(*) AS ANZAHL
FROM combined_bom
GROUP BY KOMPTEILENR, POSNR
)
WHERE ANZAHL = 1
""", {"parent_code": self.parent_code})
def _get_case_11_positions(self):
"""
Holt sich die Positionsnummern, die unter Fall 1.1 fallen.
Deren Ermittlung funktioniert so:
1. Teilenummern herausfiltern, die NICHT in der Tabelle TMP_CODES_CASE_12_OR_13 vorliegen
2. Zu den gefundenen Teilenummern folgende Spaltenwerte ermitteln und in die temporäre Tabelle TMP_POSNRS_CASE_11 schreiben:
- Teilenr des Parents
- Positionsnummer laut alter Stückliste
- Positionsnummer laut neuer Stückliste
Hinweis:
- Die Positionsnummern aus beiden Stücklisten werden benötigt, um bei der Ermittlung der Update-Daten die Zuordnung zwischen der alten und neuen
Stückliste herstellen zu können
- Die Teilenummer des Parents wird zusamment mit der Positionsnummer laut alter Stückliste dafür benötigt, um in den Tabellen STKL_REGEL und
VAR_TEIL_BEDINGUNG_2 die Zeilen zu ermitteln, die aktualisiert werden müssen
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_11")
self.db.con.execute("""
CREATE TABLE TMP_POSNRS_CASE_11 (UEBTEILENR TEXT NOT NULL,
POS_ALT INTEGER NOT NULL,
POS_NEU INTEGER NOT NULL)
""")
self.db.con.execute("""
INSERT INTO TMP_POSNRS_CASE_11 (UEBTEILENR, POS_ALT, POS_NEU)
SELECT C.UEBTEILENR, C.POSNR AS POS_ALT, RD.Positionsnummer AS POS_NEU
FROM RD_COMP RD
LEFT JOIN TMP_CODES_CASE_12_OR_13 TMP
ON RD.ChildCode = TMP.KOMPTEILENR
LEFT JOIN COMP C
ON RD.ParentCode = C.UEBTEILENR
AND RD.ChildCode = C.KOMPTEILENR
AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0)
WHERE RD.ParentCode = :parent_code
AND TMP.KOMPTEILENR IS NULL
""", {"parent_code": self.parent_code})
def _get_case_12_positions(self) -> None:
"""
Holt sich die Positionsnummern, die unter Fall 1.2 fallen.
Deren Ermittlung funktioniert so:
1. Teilenummern herausfiltern, die in der Tabelle TMP_CODES_CASE_12_OR_13, aber NICHT in Tabelle TMP_CODES_CASE_13 existieren
2. Zu den gefundenen Teilenummern folgende Spaltenwerte ermitteln und in die temporäre Tabelle TMP_POSNRS_CASE_12 schreiben:
- Teilenr des Parents
- Positionsnummer laut alter Stückliste
- Positionsnummer laut neuer Stückliste
Hinweis:
- Die Positionsnummern aus beiden Stücklisten werden benötigt, um bei der Ermittlung der Update-Daten die Zuordnung zwischen der alten und neuen
Stückliste herstellen zu können
- Die Teilenummer des Parents wird zusamment mit der Positionsnummer laut alter Stückliste dafür benötigt, um in den Tabellen STKL_REGEL und
VAR_TEIL_BEDINGUNG_2 die Zeilen zu ermitteln, die aktualisiert werden müssen
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_12")
self.db.con.execute("""
CREATE TABLE TMP_POSNRS_CASE_12 (UEBTEILENR TEXT NOT NULL,
POS_ALT INTEGER NOT NULL,
POS_NEU INTEGER NOT NULL)
""")
self.db.con.execute("""
INSERT INTO TMP_POSNRS_CASE_12 (UEBTEILENR, POS_ALT, POS_NEU)
SELECT C.UEBTEILENR, C.POSNR AS POS_ALT, RD.Positionsnummer AS POS_NEU
FROM RD_COMP RD
LEFT JOIN TMP_CODES_CASE_13 TMP
ON RD.ChildCode = TMP.KOMPTEILENR
LEFT JOIN COMP C
ON RD.ParentCode = C.UEBTEILENR
AND RD.ChildCode = C.KOMPTEILENR
AND RD.Positionsnummer = C.POSNR
WHERE RD.ParentCode = :parent_code
AND TMP.KOMPTEILENR IS NULL
AND RD.Positionsnummer NOT IN (SELECT POS_NEU FROM TMP_POSNRS_CASE_11)
""", {"parent_code": self.parent_code})
def _get_case_131_positions(self):
"""
Fall 1.31 - Positionsnummern ermitteln, die neu angelegt werden müssen.
Deren Ermittlung funktioniert so:
1. Alle Positionsnummern herausfiltern, auf die folgende Kriterien zutreffen:
- Teilenr-Längen-Kombination der Position existiert nur in der neuen Stückliste existiert
- Teilenummer der Position ist in der Tabelle TMP_CODES_CASE_13 enthalten
Hinweis:
Es reicht nicht aus, lediglich zu prüfen, ob die Teilenr-Längen-Kombination nur in der neuen Stückliste vorliegt
Grund:
Positionsnummern die unter Fall 1.2 fallen, dürfen in der neuen Stückliste eine andere Teilenr-Längen-Kombination
als in der alten Stückliste aufweisen, d.h. die Teilenr-Längen-Kombi laut neuer Stückliste liegt auch nur dort vor. Trotzdem
werden diese Positionsnummern nicht neu angelegt, sondern von der alten in die neuen Stückliste verschoben!
2. Alle gefundenen Positionen in die temporäre Tabelle TMP_POSNRS_CASE_131 schreiben (mehr Informationen werden nicht benötigt)
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_131")
self.db.con.execute("CREATE TABLE TMP_POSNRS_CASE_131 (POSNR INTEGER NOT NULL)")
self.db.con.execute("""
INSERT INTO TMP_POSNRS_CASE_131
SELECT RD.Positionsnummer AS POSNR
FROM RD_COMP RD
LEFT JOIN COMP C
ON RD.ParentCode = C.UEBTEILENR
AND RD.ChildCode = C.KOMPTEILENR
AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0)
WHERE RD.ParentCode = :parent_code
AND RD.ChildCode IN (SELECT KOMPTEILENR FROM TMP_CODES_CASE_13)
AND C.KOMPTEILENR IS NULL
""", {"parent_code": self.parent_code})
def _get_case_132_positions(self) -> None:
"""
Fall 1.32 - Positionsnummern holen, die von der alten in die neue Stückliste verschoben werden
Deren Ermittlung funktioniert so:
1. Alle Positionsnummern herausfiltern, auf die folgende Kriterien zutreffen:
- Teilenr-Längen-Kombination existiert in beiden Stücklisten
- Teilenummer der Position ist in der Tabelle TMP_CODES_CASE_13 enthalten
Hinweis:
Es reicht nicht aus, lediglich zu prüfen, ob die Teilenr-Längen-Kombination in beiden Stücklisten vorliegt
Grund:
Positionsnummern die Fall 1.1 zugeordnet werden können oder Positionsnummern, die unter Fall 1.2 fallen und deren Länge dabei in beiden
Stücklisten gleich bleibt würden grundlegend ebenfalls unter Fall 1.32 fallen und somit identifiziert werden.
Es gilt jedoch das Prioritätsprinzip, d.h. die Zuordnung zu Fall 1.1 oder Fall 1.2 erfolgt (sofern möglich) immer vor der Zuordnung
zu Fall 1.32!
2. Zu den gefundenen Teilenummern folgende Spaltenwerte ermitteln und in die temporäre Tabelle TMP_POSNRS_CASE_132 schreiben:
- Teilenr des Parents
- Positionsnummer laut alter Stückliste
- Positionsnummer laut neuer Stückliste
Hinweis:
- Die Positionsnummern aus beiden Stücklisten werden benötigt, um bei der Ermittlung der Update-Daten die Zuordnung zwischen der alten und neuen
Stückliste herstellen zu können
- Die Teilenummer des Parents wird zusamment mit der Positionsnummer laut alter Stückliste dafür benötigt, um in den Tabellen STKL_REGEL und
VAR_TEIL_BEDINGUNG_2 die Zeilen zu ermitteln, die aktualisiert werden müssen
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_132")
self.db.con.execute("""
CREATE TABLE TMP_POSNRS_CASE_132 (UEBTEILENR TEXT NOT NULL,
POS_ALT INTEGER NOT NULL,
POS_NEU INTEGER NOT NULL)
""")
self.db.con.execute("""
INSERT INTO TMP_POSNRS_CASE_132 (UEBTEILENR, POS_ALT, POS_NEU)
SELECT C.UEBTEILENR, C.POSNR AS POS_ALT, RD.Positionsnummer AS POS_NEU
FROM RD_COMP RD
INNER JOIN COMP C
ON RD.ParentCode = C.UEBTEILENR
AND RD.ChildCode = C.KOMPTEILENR
AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0)
WHERE RD.ParentCode = :parent_code
AND RD.ChildCode IN (SELECT KOMPTEILENR FROM TMP_CODES_CASE_13)
""", {"parent_code": self.parent_code})
def _get_ids_to_delete(self) -> None:
"""
Ermittelt alle Positionsnummern, die entfernt werden sollen
Hinweis:
Bei der Ermittlung dürfen die Positionsnummern, die verschoben werden NICHT berücksichtigt werden. Die alte Lösch-Logik:
"Die Teile die in der neuen Stückliste nicht drin sind, werden rausgelöscht"
ist daher nur noch eingeschränkt anwendbar!
Grund:
Das liegt vor allem an Fall 1.2, denn per Definition darf in diesem Fall die Länge in der neuen Stückliste abweichen, solange alle
Teilenr-Posnr-Kombinationen in beiden Stücklisten identsich sind, siehe hierzu z.B. Pos. 3 in Beispiel 3 unter doc/Beispiele.pdf:
Dort liegt die Länge 200 aus der alten Stückliste nicht mehr in der neuen Stückliste vor. Pos. 3 würde also laut alter Logik gelöscht werden.
Da diese Positionsnummer aber unter Fall 1.2 fällt, wird sie in die neue Stückliste verschoben, darf also nicht gelöscht werden!
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_IDS_TO_DELETE")
self.db.con.execute("""CREATE TABLE TMP_IDS_TO_DELETE (COMP_ROWID INTEGER NOT NULL,
VAR_TEIL_BED_2_ROWID INTEGER,
STKL_REGEL_ROWID INTEGER)
""")
self.db.con.execute("""
WITH COMBINED_UPDATE_DATA AS (
SELECT * FROM TMP_POSNRS_CASE_11
UNION
SELECT * FROM TMP_POSNRS_CASE_12
UNION
SELECT * FROM TMP_POSNRS_CASE_132
)
INSERT INTO TMP_IDS_TO_DELETE (COMP_ROWID, VAR_TEIL_BED_2_ROWID, STKL_REGEL_ROWID)
SELECT C.ROWID AS COMP_ROWID, B.ROWID AS VAR_TEIL_BED_2_ROWID, R.ROWID AS STKL_REGEL_ROWID
FROM COMP C
LEFT JOIN COMBINED_UPDATE_DATA CUD
ON C.UEBTEILENR = CUD.UEBTEILENR
AND C.POSNR = CUD.POS_ALT
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON C.UEBTEILENR = B.UEBTEILENR
AND C.POSNR = B.POSNR
LEFT JOIN STKL_REGEL R
ON C.UEBTEILENR = R.UEBTEILENR
AND C.POSNR = R.POSNR
WHERE C.UEBTEILENR = :parent_code
AND CUD.UEBTEILENR IS NULL
""", {"parent_code": self.parent_code})
def _get_update_data(self):
"""
Ermittlung der neuen Stücklisteninformationen, mit denen die bestehenden Datzensätze überschrieben werden
Die neuen Informationen werden für alle Positionsnummern bestimmt, die unter die Fälle 1.1, 1.2 und 1.32 fallen,
also die, die sich in den Tabellen TMP_POSNRS_CASE_11, TMP_POSNRS_CASE_12 & TMP_POSNRS_CASE_132 befinden.
Hinweis:
Es werden in diesem Zuge auch die zugehörigen Zeilen-ID's zu den Positionsnummern in den Tabellen VAR_TEIL_BEDINGUNG_2 und STKL_REGEL ermittelt,
da in diesen Tabellen ebenfalls die Positionsnummern aktualisiert werden müssen, sofern welche vorliegen. Die Regeln und Bedingungen selbst
bleiben unverändert, da sie nur verschoben werden.
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_UPDATE_DATA")
self.db.con.execute("""
CREATE TABLE TMP_UPDATE_DATA (COMP_ROWID INTEGER NOT NULL,
VAR_TEIL_BED_2_ROWID INTEGER,
STKL_REGEL_ROWID INTEGER,
POS_ALT INTEGER NOT NULL,
POS_NEU INTEGER NOT NULL,
ABW_ANZAHL INTEGER,
MENGE REAL,
LAENGE REAL,
ZUSCHNITT REAL,
DISPO_AKTIV_JN TEXT,
RUESTCODE TEXT)
""")
self.db.con.execute("""
WITH ALL_POSNRS_TO_UPDATE AS (
SELECT * FROM TMP_POSNRS_CASE_11
UNION
SELECT * FROM TMP_POSNRS_CASE_12
UNION
SELECT * FROM TMP_POSNRS_CASE_132
)
INSERT INTO TMP_UPDATE_DATA (COMP_ROWID,
VAR_TEIL_BED_2_ROWID,
STKL_REGEL_ROWID,
POS_ALT,
POS_NEU,
ABW_ANZAHL,
MENGE,
LAENGE,
ZUSCHNITT,
DISPO_AKTIV_JN,
RUESTCODE)
SELECT C.ROWID AS COMP_ROWID,
B.ROWID AS VAR_TEIL_BED_2_ROWID,
R.ROWID AS STKL_REGEL_ROWID,
P.POS_ALT,
RD.Positionsnummer AS POS_NEU,
-- Bestimmt Wert für Spalte 'ABW_ANZAHL': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV
CASE
WHEN RD.Uebergabelaenge = 0 THEN NULL
ELSE RD.Menge
END AS ABW_ANZAHL,
-- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV
CASE
WHEN RD.Uebergabelaenge = 0 THEN RD.Menge
ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT, 0)) / 1000
END AS MENGE,
-- Bestimmt Wert für Spalte 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV
CASE
WHEN RD.Uebergabelaenge = 0 THEN NULL
ELSE RD.Uebergabelaenge
END AS LAENGE,
C.ZUSCHNITT,
C.DISPO_AKTIV_JN,
C.RUESTCODE
FROM ALL_POSNRS_TO_UPDATE P
INNER JOIN COMP C
ON P.UEBTEILENR = C.UEBTEILENR
AND P.POS_ALT = C.POSNR
INNER JOIN RD_COMP RD
ON P.UEBTEILENR = RD.ParentCode
AND P.POS_NEU = RD.Positionsnummer
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON P.UEBTEILENR = B.UEBTEILENR
AND P.POS_ALT = B.POSNR
LEFT JOIN STKL_REGEL R
ON P.UEBTEILENR = R.UEBTEILENR
AND P.POS_ALT = R.POSNR
""")
def delete(self):
"""
- Entfernt die Positionsnummern aus COMP, die nicht mehr in der neuen Stückliste vorliegen
- Entfernt die TOS-Regeln und -Bedingungen der Positionsnummern aus VAR_TEIL_BEDINGUNG_2 und STKL_REGEL
"""
# Stücklistenpositionen entfernen
self.db.con.execute("DELETE FROM COMP WHERE ROWID IN (SELECT COMP_ROWID FROM TMP_IDS_TO_DELETE)")
# TOS-Bedingungen entfernen
self.db.con.execute("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE ROWID IN (SELECT VAR_TEIL_BED_2_ROWID FROM TMP_IDS_TO_DELETE)")
# TOS-Regeln entfernen
self.db.con.execute("DELETE FROM STKL_REGEL WHERE ROWID IN (SELECT STKL_REGEL_ROWID FROM TMP_IDS_TO_DELETE)")
def update(self):
"""
Aktualisiert die Positionsnummern, deren Spaltenanpassungen und TOS-Regeln / -Bedingungen von der alten in die neue Stückliste übernommen werden sollen.
"""
# Stücklisteninformationen updaten
self.db.con.execute("""
UPDATE COMP
SET
POSNR = UD.POS_NEU,
ABW_ANZAHL = UD.ABW_ANZAHL,
MENGE = UD.MENGE,
LAENGE = UD.LAENGE,
ZUSCHNITT = UD.ZUSCHNITT,
DISPO_AKTIV_JN = UD.DISPO_AKTIV_JN,
RUESTCODE = UD.RUESTCODE
FROM TMP_UPDATE_DATA UD
WHERE ROWID = UD.COMP_ROWID
""")
# TOS-Bedingungen updaten
self.db.con.execute("""
UPDATE VAR_TEIL_BEDINGUNG_2
SET POSNR = UD.POS_NEU
FROM TMP_UPDATE_DATA UD
WHERE ROWID = UD.VAR_TEIL_BED_2_ROWID
""")
# TOS-Regeln updaten
self.db.con.execute("""
UPDATE STKL_REGEL
SET POSNR = UD.POS_NEU
FROM TMP_UPDATE_DATA UD
WHERE ROWID = UD.STKL_REGEL_ROWID
""")
def insert(self):
"""
Neuanlage aller Positionen, die nur in der neuen Stückliste vorliegen (das sind die aus der Tabelle TMP_POSNRS_CASE_131)
Hinweise:
- Neue Positionen werden immer ohne TOS-Regeln oder -Bedingungen angelegt, d.h. es erfolgen keine INSERT-Statements
in den Tabellen VAR_TEIL_BEDINGUNG_2 oder STKL_REGEL!
- Zur Verdeutlichung, welche Werte z.B. in der Spalte Ruestcode oder Dispoaktiv nach der Neuanlage vorliegen, werden
die Defaultwerte aus der PARTS-Tabelle geholt. Ob ein solcher INSERT auch in SIVAS erforderlich ist, war zum Zeitpunkt
der Erstellung dieses Scripts nicht klar.
"""
self.db.con.execute("""
INSERT INTO COMP (UEBTEILENR, POSNR, KOMPTEILENR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE)
SELECT RD.ParentCode AS UEBTEILENR,
RD.Positionsnummer AS POSNR,
RD.ChildCode AS KOMPTEILENR,
-- Bestimmt Wert für Spalte 'ABW_ANZAHL': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV
CASE
WHEN RD.Uebergabelaenge = 0 THEN NULL
ELSE RD.Menge
END AS ABW_ANZAHL,
-- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV
CASE
WHEN RD.Uebergabelaenge = 0 THEN RD.Menge
ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT, 0)) / 1000
END AS MENGE,
-- Bestimmt Wert für Spalte 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV
CASE
WHEN RD.Uebergabelaenge = 0 THEN NULL
ELSE RD.Uebergabelaenge
END AS LAENGE,
NULL AS ZUSCHNITT,
P.DISPO_AKTIV_JN,
P.RUESTCODE
FROM RD_COMP RD
LEFT JOIN COMP C
ON RD.ChildCode = C.KOMPTEILENR
AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0)
AND RD.ParentCode = C.UEBTEILENR
LEFT JOIN PARTS P
ON RD.ChildCode = P.TEILENR
WHERE C.KOMPTEILENR IS NULL
AND RD.ParentCode = :parent_code
AND RD.Positionsnummer IN (SELECT POSNR FROM TMP_POSNRS_CASE_131)
""", {"parent_code": self.parent_code})
def drop_tmp_tables(self):
self.db.con.execute("DROP TABLE TMP_CODES_CASE_12_OR_13")
self.db.con.execute("DROP TABLE TMP_CODES_CASE_13")
self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_11")
self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_12")
self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_131")
self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_132")
self.db.con.execute("DROP TABLE TMP_IDS_TO_DELETE")
self.db.con.execute("DROP TABLE TMP_UPDATE_DATA")
def write_new_bom(self) -> None:
self._get_codes_case_12_or_13()
self._get_codes_case_13()
self._get_case_11_positions()
self._get_case_12_positions()
self._get_case_131_positions()
self._get_case_132_positions()
self._get_ids_to_delete()
self._get_update_data()
self.delete()
self.insert()
self.update()
self.drop_tmp_tables()
class AVWorkflowMessage:
def __init__(self, db: Database, parent_code: str):
self.db = db
self.parent_code = parent_code
def __str__(self):
res = self.get_data()
msg = self._create_msg(res)
return msg
def drop_tmp_table(self):
self.db.con.execute("DROP TABLE TMP_AV_WORKFLOW_MESSAGE")
def get_av_workflow_posnrs_and_codes(self):
"""
Wenn Positionsnummern folgende Kriterien erfüllen, sollen Sie in einer Workflowmeldung an die AV-Verantwortlichen ausgegeben werden:
Kriterien:
WENN die Positionen zu einer längenabhängigen Teilenummer MIT Sägeliste gehören (d.h. Haken bei 'Halbzeug ohne Sägeliste' im Teilestamm ist nicht gesetzt)
UND diese längenabhängige Teilenummer in der alten und der neuen Stückliste vertreten ist
UND die Länge-Teilenummer-Kombination dieser Positionsnummer nur in der alten Stückliste vorliegt (aber nicht in der Neuen)
UND bei der Position mindestens eine der nachfolgenden Kriterien erfüllt ist:
- der Wert in der Spalte 'Rüstcode' weicht vom Rüstcode-Wert ab, den die zugehörige Teilenummer im Teilstamm aufweist
- in der Spalte 'Dispoaktiv' liegt kein Haken vor
- in der Spalte 'Zuschnitt' ist ein Wert definiert
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_AV_WORKFLOW_MESSAGE")
self.db.con.execute("""
CREATE TABLE TMP_AV_WORKFLOW_MESSAGE (POSNR INTEGER,
KOMPTEILENR TEXT)
""")
test = self.db.con.execute("""
--längenabhängige Teilenummern ermitteln, die in beiden Stücklisten vorliegen
WITH laengenabh_codes_in_both_boms AS (
SELECT DISTINCT ChildCode
FROM RD_COMP
WHERE ParentCode = :parent_code
AND ChildCode in (
SELECT DISTINCT KOMPTEILENR
FROM COMP
WHERE UEBTEILENR = :parent_code
AND COALESCE(LAENGE, 0) != 0
)
)
INSERT INTO TMP_AV_WORKFLOW_MESSAGE (POSNR, KOMPTEILENR)
SELECT C.POSNR, C.KOMPTEILENR
FROM COMP C
LEFT JOIN RD_COMP RD
ON C.UEBTEILENR = RD.ParentCode
AND C.KOMPTEILENR = RD.ChildCode
AND COALESCE(C.LAENGE, 0) = RD.Uebergabelaenge
INNER JOIN PARTS P
ON C.KOMPTEILENR = P.TEILENR
WHERE C.UEBTEILENR = :parent_code
AND C.KOMPTEILENR IN laengenabh_codes_in_both_boms -- Nur längenabhängige Teilenummern berücksichtigen, die in beiden Stücklisten vorliegen
AND RD.ParentCode IS NULL -- Nur Positionen berücksichtigen, deren Länge ausschließlich in der alten Stückliste vorliegt
AND P.INFO2_JN = "N" -- Nur Positionen berücksichtigen, deren zugehörige Teilenummer keinen Haken bei "Halbzeug ohne Sägeliste" im Teilestamm aufweist
AND ( -- Nur Positionen berücksichtigen, auf die min. eines der nachfolgenden Kriterien zutrifft:
C.ZUSCHNITT IS NOT NULL -- Zuschnittswert liegt vor
OR C.RUESTCODE != P.RUESTCODE -- Rüstcode der Position in der Stammstückliste ist abweichend zum Rüstcode der zugehörigen Teilenummer im Teilestamm
OR C.DISPO_AKTIV_JN = "N" -- Stücklistenpositionen besitzt keinen Haken in der Spalte "Dispoaktiv"
)
ORDER BY C.POSNR
""", {"parent_code": self.parent_code}).fetchall()
return test
def get_data(self) -> list[dict[str]]:
res = self.db.con.execute("SELECT POSNR, KOMPTEILENR FROM TMP_AV_WORKFLOW_MESSAGE").fetchall()
return res
def _create_msg(self, data: list[dict[str]] | None) -> str:
if not data:
return "\nAV-Meldung:\nKeine"
affected_positions_str = ", ".join([f"{pos["POSNR"]} - {pos["KOMPTEILENR"]}" for pos in data])
return f"\nAV-Meldung:\nBitte Stückliste {self.parent_code} prüfen. Vor der Änderung lagen bei folgenden Positionen manuelle Eintragungen / Änderungen in den Spalten „Zuschnitt“, „Rüstcode“ oder „Dispoaktiv“ vor: {affected_positions_str}"
class TOSWorkflowMessage:
def __init__(self, db: Database, parent_code: str):
self.db = db
self.parent_code = parent_code
def __str__(self):
res = self.get_data()
msg = self._create_msg(res)
return msg
def drop_tmp_table(self):
self.db.con.execute("DROP TABLE TMP_TOS_WORKFLOW_MESSAGE")
def get_tos_workflow_posnrs_and_codes(self) -> None:
"""
Wenn Positionsnummern in der Stückliste TOS-Regeln oder -Bedingungen aufweisen,
sollen Sie in einer Workflowmeldung an die TOS-Verantwortlichen ausgegeben werden.
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_TOS_WORKFLOW_MESSAGE")
self.db.con.execute("""
CREATE TABLE TMP_TOS_WORKFLOW_MESSAGE (POSNR INTEGER,
KOMPTEILENR TEXT)
""")
self.db.con.execute("""
INSERT INTO TMP_TOS_WORKFLOW_MESSAGE (POSNR, KOMPTEILENR)
SELECT C.POSNR, C.KOMPTEILENR
FROM COMP C
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON C.UEBTEILENR = B.UEBTEILENR
AND C.POSNR = B.POSNR
LEFT JOIN STKL_REGEL R
ON C.UEBTEILENR = R.UEBTEILENR
AND C.POSNR = R.POSNR
WHERE C.UEBTEILENR = :parent_code
AND (
B.UEB_LEISTEN_NR IS NOT NULL
OR R.LFD_NR IS NOT NULL
)
ORDER BY C.POSNR
""", {"parent_code": self.parent_code})
def get_data(self) -> list[dict[str]]:
res = self.db.con.execute("SELECT POSNR, KOMPTEILENR FROM TMP_TOS_WORKFLOW_MESSAGE").fetchall()
return res
def _create_msg(self, data: list[dict[str]] | None) -> str:
if not data:
return "\nTOS-Meldung:\nKeine"
affected_positions_str = ", ".join([f"{pos["POSNR"]} - {pos["KOMPTEILENR"]}" for pos in data])
return f"\nTOS-Meldung:\nBitte Stückliste {self.parent_code} prüfen. Vor der Änderung lagen bei folgenden Positionen TOS-Bedingungen vor: {affected_positions_str}"
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink