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

769 lines
38 KiB
Python
Raw Blame History

import argparse
import csv
import os
import sqlite3
import subprocess
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")
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 fertig bearbeiteten Stückliste in der Konsole")
sp_csv = subparsers.add_parser("csv", help="Ausgabe der fertig bearbeiteten Stückliste als CSV-Datei")
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")
sp_csv.add_argument("-o", "--open_csv", action="store_true", help="CSV-Datei nach dem schreiben öffnen")
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"])
# 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)
# Neue Stückliste erstellen / schreiben
bom_processing.write_new_bom()
if args["output_type"] == "cmd":
if args["verbose"]:
output = repr(new_bom)
print(output)
else:
print(new_bom)
elif args["output_type"] == "csv":
if args["verbose"]:
bom_data = new_bom.get_verbose_data()
else:
bom_data = new_bom.get_compact_data()
filedir = args["out_dir"]
filename = f"BOM_{args["parent_code"]}_final.csv"
filepath = os.path.join(filedir, filename)
csv_handler.write_from_dict(data=bom_data, filedir=filedir, filename=filename)
if args["open_csv"]:
subprocess.run(['cmd', '/c', 'start', '', filepath], shell=True)
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.REGEL, B.BEDINGUNG
FROM COMP C
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON C.ID = B.COMP_ID
LEFT JOIN STKL_REGEL R
ON C.ID = R.COMP_ID
WHERE C.PARENT = :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.REGEL, B.Bedingung
FROM COMP C
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON C.ID = B.COMP_ID
LEFT JOIN STKL_REGEL R
ON C.ID = R.COMP_ID
WHERE C.PARENT = :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, 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 deutlich komplexer wird.
Die Prüfung läuft so ab:
1. Teilenr-Längen-Kombinationen aus der alten und neuen Stückliste mit UNION ALL zusammenfügen
2. Mit COUNT ermitteln, wie oft jede Teilenummer-Längen-Kombination in der zusammengefügten Stückliste auftritt
3. Die Teilenummern in die temporäre Tabelle TMP_CODES_CASE_12_OR_13 schreiben, deren Teilenr-Längen-Kombination 1x vorkommen
=> 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 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, PRIMARY KEY (KOMPTEILENR))")
self.db.con.execute("""
WITH combined_bom AS (
SELECT C.KOMPTEILENR, COALESCE(C.LAENGE, 0) AS LAENGE
FROM COMP C
WHERE C.PARENT = :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
sie Fall 1.32 auf diese Positionen zutreffen würde (Vorrangsprinzip).
Die Prüfung läuft so ab:
1. Teilenr-Posnr-Kombinationen aus der alten und neuen Stückliste mit UNION ALL zusammenfügen, 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 ausgeben, deren Teilenr-Posnr-Kombination nur 1x vorkommen 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 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, PRIMARY KEY(KOMPTEILENR))")
self.db.con.execute("""
WITH combined_bom AS (
SELECT KOMPTEILENR, POSNR
FROM COMP
WHERE PARENT = :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:
- Parent-Teilenr
- Posnr. laut neuer Stückliste
- ID laut alter Stückliste
in die temporäre Tabelle TMP_POSNRS_CASE_11 schreiben.
Hinweis:
- ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen
- Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_11")
self.db.con.execute("""CREATE TABLE TMP_POSNRS_CASE_11 (COMP_ID INTEGER NOT NULL,
PARENT TEXT NOT NULL,
POSNR INTEGER NOT NULL,
PRIMARY KEY(POSNR))""")
self.db.con.execute("""
INSERT INTO TMP_POSNRS_CASE_11 (COMP_ID, PARENT, POSNR)
SELECT C.ID AS COMP_ID, C.PARENT, RD.Positionsnummer AS POSNR
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.PARENT
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:
- Parent-Teilenr
- Posnr. laut neuer Stückliste
- ID laut alter Stückliste
in die temporäre Tabelle TMP_POSNRS_CASE_12 schreiben.
Hinweis:
- ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen
- Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_12")
self.db.con.execute("""
CREATE TABLE TMP_POSNRS_CASE_12 (COMP_ID INTEGER NOT NULL,
PARENT TEXT NOT NULL,
POSNR INTEGER NOT NULL,
PRIMARY KEY(POSNR))
""")
self.db.con.execute("""
INSERT INTO TMP_POSNRS_CASE_12 (COMP_ID, PARENT, POSNR)
SELECT C.ID AS COMP_ID, C.PARENT, RD.Positionsnummer AS POSNR
FROM RD_COMP RD
LEFT JOIN TMP_CODES_CASE_13 TMP
ON RD.ChildCode = TMP.KOMPTEILENR
LEFT JOIN COMP C
ON RD.ParentCode = C.PARENT
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 POSNR FROM TMP_POSNRS_CASE_11);
""", {"parent_code": self.parent_code})
def _get_case_131_positions(self):
"""
Fall 1.31 - Positionsnummern holen, die neu angelegt werden müssen.
Deren Ermittlung funktioniert so:
1. Teilenummern herausfiltern, deren Teilenr-Längen-Kombination nur in der neuen Stückliste existieren und die zudem in der Tabelle
TMP_CODES_CASE_13 vorliegen.
Hinweis:
Es reicht nicht aus, lediglich zu prüfen, ob die Teilenr-Längen-Kombination nur in der neuen Stücklist 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. Zu den gefundenen Teilenummern:
- Parent-Teilenr
- Posnr. laut neuer Stückliste
- ID laut alter Stückliste
in die temporäre Tabelle TMP_POSNRS_CASE_131 schreiben.
Hinweis:
- ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen
- Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können
"""
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, PRIMARY KEY(POSNR))")
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.PARENT
AND RD.ChildCode = C.KOMPTEILENR
AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0)
WHERE RD.ParentCode = :parent_code
AND RD.ChildCode IN (SELECT * 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. Teilenummern herausfiltern, deren Teilenr-Längen-Kombination in beiden Stückliste existieren und die zudem in der Tabelle
TMP_CODES_CASE_13 vorliegen. Lediglich zu prüfen, ob die Teilenr-Längen-Kombination nur in der neuen Stückliste vorliegt reicht nicht aus!
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 Vorrangsprinzip, 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:
- Parent-Teilenr
- Posnr. laut neuer Stückliste
- ID laut alter Stückliste
in die temporäre Tabelle TMP_POSNRS_CASE_132 schreiben.
Hinweis:
- ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen
- Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_132")
self.db.con.execute("""
CREATE TABLE TMP_POSNRS_CASE_132 (COMP_ID INTEGER NOT NULL,
PARENT TEXT NOT NULL,
POSNR INTEGER NOT NULL,
PRIMARY KEY(POSNR))""")
self.db.con.execute("""
INSERT INTO TMP_POSNRS_CASE_132 (COMP_ID, PARENT, POSNR)
SELECT C.ID AS COMP_ID, C.PARENT, RD.Positionsnummer AS POSNR
FROM RD_COMP RD
INNER JOIN COMP C
ON RD.ParentCode = C.PARENT
AND RD.ChildCode = C.KOMPTEILENR
AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0)
WHERE RD.ParentCode = :parent_code
AND RD.ChildCode IN (SELECT * FROM TMP_CODES_CASE_13)
""", {"parent_code": self.parent_code})
def delete(self):
"""
Entfernt die Positionsnummern aus der alten Stückliste, die nicht mehr in der neuen Stückliste vorliegen.
Hinweis:
Bei der Ermittlung der zu löschenden Datensätze 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" 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, d.h. sie darf 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 (ID INTEGER NOT NULL, PRIMARY KEY(ID))")
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
SELECT C.ID
FROM COMP C
LEFT JOIN COMBINED_UPDATE_DATA CUD
ON C.ID = CUD.COMP_ID
WHERE C.PARENT = :parent_code
AND CUD.COMP_ID IS NULL
""", {"parent_code": self.parent_code})
# Entfernung der Stücklistenpositionen aus COMP
self.db.con.execute("DELETE FROM COMP WHERE ID IN (SELECT ID FROM TMP_IDS_TO_DELETE)")
# Entfernung der TOS-Bedingungen aus VAR_TEIL_BEDINGUNG_2
self.db.con.execute("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE COMP_ID IN (SELECT ID FROM TMP_IDS_TO_DELETE)")
# Entfernung der TOS-Regeln aus STKL_REGEL
self.db.con.execute("DELETE FROM STKL_REGEL WHERE COMP_ID IN (SELECT ID FROM TMP_IDS_TO_DELETE)")
def update(self):
"""
Aktualisiert die Positionsnummern, die von der alten in die neue Stückliste verschoben werden sollen.
Dies umfasst alle Positionsnummern, die unter die Fälle 1.1, 1.2 und 1.32 fallen. Sie wurden in den Methoden
_get_case_11_positions, _get_case_12_positions & _get_case_132_positions ermittelt und in die temporären
Tabellen TMP_POSNRS_CASE_11, TMP_POSNRS_CASE_12, TMP_POSNRS_CASE_132 geschrieben.
Jetzt passiert folgendes:
1. Die Positionen aus den 3 Tabellen werden zu einer temporären Tabelle, ALL_POSNRS_TO_UPDATE kombiniert.
2. Zu allen Positionen aus in dieser kombinierten, temporären Tabelle werden dann die neuen Stücklisteninformationen aus COMP,
bzw. RD_COMP bestimmt und in eine weitere, temporäre, aber persistente Tabelle TMP_UPDATE_DATA geschrieben.
3. Durchführen der Updates in den Tabellen COMP (für die Stücklisteninformationen), VAR_TEIL_BEDINGUNG_2 (für die TOS-Bedingungen)
und STKL_REGEL für die TOS-Regeln
"""
self.db.con.execute("DROP TABLE IF EXISTS TMP_UPDATE_DATA")
self.db.con.execute("CREATE TABLE TMP_UPDATE_DATA (ID, POSNR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE, BEDINGUNG, REGEL)")
# Neue Stücklisteninformationen ermitteln
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 (ID, POSNR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE, BEDINGUNG, REGEL)
SELECT C.ID,
RD.Positionsnummer AS POSNR,
-- 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,
B.BEDINGUNG,
R.REGEL
FROM ALL_POSNRS_TO_UPDATE CUD
INNER JOIN COMP C
ON CUD.COMP_ID = C.ID
INNER JOIN RD_COMP RD
ON CUD.PARENT = RD.ParentCode
AND CUD.POSNR = RD.Positionsnummer
LEFT JOIN VAR_TEIL_BEDINGUNG_2 B
ON CUD.COMP_ID = B.COMP_ID
LEFT JOIN STKL_REGEL R
ON CUD.COMP_ID = R.COMP_ID
""")
# Stücklisteninformationen updaten
self.db.con.execute("""
UPDATE COMP
SET POSNR = UD.POSNR,
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 COMP.ID = UD.ID
""")
# TOS-Bedingungen updaten
self.db.con.execute("""
UPDATE VAR_TEIL_BEDINGUNG_2
SET BEDINGUNG = UD.BEDINGUNG
FROM TMP_UPDATE_DATA UD
WHERE COMP_ID = UD.ID
""")
# TOS-Regeln updaten
self.db.con.execute("""
UPDATE STKL_REGEL
SET REGEL = UD.REGEL
FROM TMP_UPDATE_DATA UD
WHERE COMP_ID = UD.ID
""")
def insert(self):
"""
Neuanlage aller Positionen, die nur in der neuen Stückliste vorliegen.
Die Positionsnummern wurde bereits in der Methode _get_codes_case_131 ermittelt und in die temporäre Tabelle
TMP_POSNRS_CASE_131 geschrieben.
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 nicht klar.
"""
self.db.con.execute("""
INSERT INTO COMP (PARENT, POSNR, KOMPTEILENR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE)
SELECT RD.ParentCode AS PARENT,
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.PARENT
LEFT JOIN PARTS P
ON RD.ChildCode = P.TEILENR
WHERE C.KOMPTEILENR IS NULL
AND RD.ParentCode = :parent_code
AND RD.Positionsnummer IN (SELECT * 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.delete()
self.insert()
self.update()
self.drop_tmp_tables()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink