Komplett überarbeitet und in Klassen aufgeteilt, um Verantwortlichkeiten besser aufteilen zu können. Beachte - Die Verarbeitungslogik der neuen Stückliste (die nun in der Klasse BOMProcessing und hauptsächlich in der darin enthaltenen Methode 'write_new_bom()' befindet) hat sich durch das Refactoring nicht geändert!
This commit is contained in:
+467
-281
@@ -2,78 +2,361 @@ import argparse
|
||||
import csv
|
||||
import os
|
||||
import sqlite3
|
||||
from contextlib import closing
|
||||
|
||||
def main(asm_partnum, sql_filedir, csv_filedir):
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='Programm bearbeitet Stücklisten-CSV aus RuleDesigner, sodass Spaltenüberschreibungen + TOS-Bedingungen und -Regeln (falls vorhanden) mit verschoben werden',
|
||||
prog='verschiebelogik',
|
||||
formatter_class=lambda prog: argparse.HelpFormatter(prog, max_help_position=50))
|
||||
parser.register("action", "store_after_check", ArgparseDirectoryCheck)
|
||||
parser.add_argument("parent_code", action="store", type=str, help="Teilenummer der Baugruppe")
|
||||
parser.add_argument("-cid", "--in_csv_dir", action="store_after_check", default=os.environ.get("PROJECT_DATA"), help="Eingabeverzeichnis der Stücklisten-CSV aus RuleDesigner")
|
||||
subparsers = parser.add_subparsers(title="output type", help="Ausgabe der fertigen Stückliste in Konsole oder als CSV", dest="output_type")
|
||||
sp_cmd = subparsers.add_parser("console", 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.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben")
|
||||
sp_csv.register("action", "store_after_check", ArgparseDirectoryCheck)
|
||||
sp_cmd.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben")
|
||||
sp_csv.add_argument("-cod", "--out_csv_dir", action="store_after_check", default=os.environ.get("PROJECT_WORK"), help="Ausgabeverzeichnis fertigen Stücklisten-CSV")
|
||||
args: dict = vars(parser.parse_args())
|
||||
|
||||
build_sivas_from_file(sql_filedir)
|
||||
csv_data = read_csv(csv_filedir, asm_partnum)
|
||||
csv_to_db(csv_data)
|
||||
create_new_bom(asm_partnum)
|
||||
# print_new_bom(asm_partnum)
|
||||
write_bom_as_csv(asm_partnum)
|
||||
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!")
|
||||
|
||||
def csv_to_db(csv_data: list[dict[str, str]]) -> None:
|
||||
"""Schreibt neue Stücklistendaten aus CSV in temporäre Datenbanktabelle."""
|
||||
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"])
|
||||
|
||||
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))
|
||||
""")
|
||||
# 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)
|
||||
|
||||
for line in csv_data:
|
||||
con.execute(""" INSERT INTO RD_COMP (ParentCode, Positionsnummer, ChildCode, Menge, Uebergabelaenge)
|
||||
VALUES (:ParentCode, :Positionsnummer, :ChildCode, :Menge, :Uebergabelaenge)""", line)
|
||||
# CSV-Daten in Datenbank schreiben
|
||||
csv_data = csv_handler.read_as_dict(filename=f"BOM_{args["parent_code"]}")
|
||||
new_bom.create()
|
||||
new_bom.insert_csv_data(csv_data)
|
||||
|
||||
def dict_factory(cursor, row):
|
||||
# Neue Stückliste erstellen / schreiben
|
||||
bom_processing.write_new_bom()
|
||||
|
||||
if args["output_type"] == "console":
|
||||
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()
|
||||
|
||||
csv_handler.write_from_dict(data=bom_data,
|
||||
filename=f"BOM_{args["parent_code"]}_final",
|
||||
filedir=args["out_csv_dir"])
|
||||
else:
|
||||
NotImplementedError
|
||||
|
||||
finally:
|
||||
db.close()
|
||||
db.remove() # Damit bei jedem Lauf eine neue erstellt 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 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"), ) -> 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_WORK")) -> 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)
|
||||
|
||||
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")) -> 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 read_csv(filedir: str, partnumber: str) -> list[dict[str, str]]:
|
||||
csv_path = os.path.join(filedir, f"BOM_{partnumber}.csv")
|
||||
if not csv_path:
|
||||
raise FileNotFoundError
|
||||
def close(self):
|
||||
self.con.close()
|
||||
|
||||
with open(csv_path, "r", encoding="utf-8", newline="") as csv_file:
|
||||
reader = csv.DictReader(csv_file, delimiter=";")
|
||||
csv_data = []
|
||||
for line in reader:
|
||||
line["Positionsnummer"] += "0" # Angleichung Positionsnummer an SIVAS-Format (Pos. * 10)
|
||||
csv_data.append(line)
|
||||
def remove(self):
|
||||
if os.path.isfile(self.db_path):
|
||||
os.remove(self.db_path)
|
||||
|
||||
return csv_data
|
||||
|
||||
def build_sivas_from_file(sql_filedir: str, sql_filename="sivas_replika.sql") -> None:
|
||||
filepath = os.path.join(sql_filedir, sql_filename)
|
||||
class RD_COMP:
|
||||
|
||||
if not filepath:
|
||||
raise FileNotFoundError
|
||||
"""
|
||||
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.
|
||||
"""
|
||||
|
||||
with open(filepath, "r", encoding="utf-8", newline="") as sql_file:
|
||||
sql = sql_file.read()
|
||||
def __init__(self, db: Database, parent_code: str):
|
||||
self.db = db
|
||||
self.parent_code = parent_code
|
||||
|
||||
con.executescript(sql)
|
||||
def __repr__(self):
|
||||
res = self.get_verbose_data()
|
||||
output = self._sql_as_ascii_table(res)
|
||||
return output
|
||||
|
||||
def get_new_bom_data(parent_partnum:str) -> list[dict[str, str]]:
|
||||
""" Positionsnummern und zugehörige Teilenummern aus neuer Stückliste holen"""
|
||||
res = con.execute("SELECT * FROM RD_COMP WHERE ParentCode = :parent ORDER BY Positionsnummer", {"parent": parent_partnum}).fetchall()
|
||||
def __str__(self):
|
||||
res = self.get_compact_data()
|
||||
output = self._sql_as_ascii_table(res)
|
||||
return output
|
||||
|
||||
|
||||
def create(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):
|
||||
pass
|
||||
|
||||
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_all_bom_data(self) -> list[dict[str]]:
|
||||
res = self.db.con.execute("SELECT * FROM RD_COMP WHERE ParentCode = :parent ORDER BY Positionsnummer",
|
||||
{"parent": self.parent_code}).fetchall()
|
||||
return res
|
||||
|
||||
def get_old_bom_ids(parent_partnum: str) -> list | None:
|
||||
"""Gibt ID's der Datensätze aus der alten Stückliste zurück"""
|
||||
res = con.execute("SELECT ID FROM COMP WHERE PARENT = :parent", {"parent": parent_partnum}).fetchall()
|
||||
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_positions_to_move(parent_partnum: str) -> set | None:
|
||||
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]]):
|
||||
"""SQL-Rohdaten 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)
|
||||
|
||||
def insert(self, new_data: dict[str, str]) -> None:
|
||||
self.db.con.executemany("""INSERT INTO COMP (PARENT, POSNR, KOMPTEILENR, DISPO_AKTIV_JN, MENGE, ABW_ANZAHL, LAENGE, ZUSCHNITT, RUESTCODE)
|
||||
VALUES (:PARENT, :POSNR, :KOMPTEILENR, :DISPO_AKTIV_JN, :MENGE, :ABW_ANZAHL, :LAENGE, :ZUSCHNITT, :RUESTCODE)""",
|
||||
new_data)
|
||||
|
||||
def delete(self, ids: list[dict[str]]) -> None:
|
||||
self.db.con.executemany("DELETE FROM COMP WHERE ID = :ID", ids)
|
||||
|
||||
def update(self, pos_data: list[dict[str]]) -> None:
|
||||
self.db.con.executemany("""
|
||||
UPDATE COMP
|
||||
SET POSNR = :POSNR,
|
||||
ABW_ANZAHL = :ABW_ANZAHL,
|
||||
MENGE = :MENGE,
|
||||
LAENGE = :LAENGE,
|
||||
ZUSCHNITT = :ZUSCHNITT,
|
||||
DISPO_AKTIV_JN = :DISPO_AKTIV_JN,
|
||||
RUESTCODE = :RUESTCODE
|
||||
WHERE ID = :ID
|
||||
""", pos_data)
|
||||
|
||||
|
||||
class COMP:
|
||||
|
||||
"""Beinhaltet Methoden, zur Interaktion mit Tabelle COMP (gleichnamig wie in SIVAS)"""
|
||||
|
||||
def __init__(self, db: Database):
|
||||
self.db = db
|
||||
|
||||
def rebuild_sivas(self, sql: str) -> None:
|
||||
self.db.con.executescript(sql)
|
||||
|
||||
def get_child_ids(self, parent_code: str) -> list[dict[str]]:
|
||||
res = self.db.con.execute("SELECT ID FROM COMP WHERE PARENT = :parent",
|
||||
{"parent": parent_code}).fetchall()
|
||||
return res
|
||||
|
||||
|
||||
class VAR_TEIL_BEDINGUNGEN_2:
|
||||
|
||||
"""
|
||||
Beinhaltet Methoden, zur Interaktion mit Tabelle VAR_TEIL_BEDINGUNG_2 (gleichnamig wie in SIVAS),
|
||||
in der TOS-Bedingungen gespeichert sind
|
||||
"""
|
||||
|
||||
def __init__(self, db: Database) -> None:
|
||||
self.db = db
|
||||
|
||||
def update(self, data: list[dict[str]]) -> None:
|
||||
self.db.con.executemany("UPDATE VAR_TEIL_BEDINGUNG_2 SET BEDINGUNG = :BEDINGUNG WHERE COMP_ID = :ID", data)
|
||||
|
||||
def delete(self, data: list[dict[str]]) -> None:
|
||||
self.db.con.executemany("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE COMP_ID = :ID", data)
|
||||
|
||||
|
||||
class STKL_REGEL:
|
||||
|
||||
"""
|
||||
Beinhaltet Methoden, zur Interaktion mit Tabelle STKL_REGEL (gleichnamig wie in SIVAS),
|
||||
in der TOS-Regeln gespeichert sind
|
||||
"""
|
||||
|
||||
def __init__(self, db: Database) -> None:
|
||||
self.db = db
|
||||
|
||||
def update(self, data: list[dict[str]]) -> None:
|
||||
self.db.con.executemany("UPDATE STKL_REGEL SET REGEL = :REGEL WHERE COMP_ID = :ID", data)
|
||||
|
||||
def delete(self, data: list[dict[str]]) -> None:
|
||||
self.db.con.executemany("DELETE FROM STKL_REGEL WHERE COMP_ID = :ID", data)
|
||||
|
||||
class BOMProcessing:
|
||||
|
||||
"""
|
||||
Die Methoden dieser Klasse verarbeiten die Daten aus der RuleDesigner-Stücklisten CSV und initiiert das Schreiben der verarbeiteten
|
||||
Daten in COMP.
|
||||
"""
|
||||
|
||||
def __init__(self, db: Database, parent_code: str):
|
||||
self.db = db
|
||||
self.parent_code = parent_code
|
||||
self.new_bom = RD_COMP(db, parent_code)
|
||||
self.old_bom = COMP(db)
|
||||
self.tos_bedingungen = VAR_TEIL_BEDINGUNGEN_2(db)
|
||||
self.tos_regeln = STKL_REGEL(db)
|
||||
|
||||
def extract_col_data(self, query_results: list[dict[str]], *col_names: tuple) -> list[dict[str]]:
|
||||
extracted_data = []
|
||||
for result in query_results:
|
||||
data = {col_name: result[col_name] for col_name in col_names if col_name in result}
|
||||
extracted_data.append(data)
|
||||
return extracted_data
|
||||
|
||||
def _get_positions_to_move_old_logic(self) -> set | None:
|
||||
"""
|
||||
Gibt alle Positionsnummern zurück, die gemäß alter Logik verschoben werden
|
||||
"""
|
||||
res = con.execute("""
|
||||
res = self.db.con.execute("""
|
||||
SELECT RD.Positionsnummer AS POS
|
||||
|
||||
FROM RD_COMP RD
|
||||
@@ -83,17 +366,17 @@ def get_positions_to_move(parent_partnum: str) -> set | None:
|
||||
AND RD.Uebergabelaenge = COALESCE(C.LAENGE,0)
|
||||
|
||||
WHERE RD.ParentCode = :parent
|
||||
""",
|
||||
{"parent": parent_partnum}).fetchall()
|
||||
""", {"parent": self.parent_code}).fetchall()
|
||||
|
||||
if res is not None:
|
||||
res = set([line["POS"] for line in res])
|
||||
return res
|
||||
|
||||
def identical_occurences_in_both_boms(parent_partnum: str, child_id: str) -> bool:
|
||||
def _check_identical_code_len_combo(self, child_code: str) -> bool:
|
||||
"""
|
||||
Prüfung Fall 1.1:
|
||||
"""
|
||||
res = con.execute("""
|
||||
res = self.db.con.execute("""
|
||||
SELECT CASE
|
||||
-- Prüfung 1: Liegen alle Teilenr.+Länge-Kombinationen aus der neuen Stückliste auch in der alten Stückliste vor?
|
||||
WHEN NOT EXISTS (
|
||||
@@ -126,15 +409,15 @@ def identical_occurences_in_both_boms(parent_partnum: str, child_id: str) -> boo
|
||||
ELSE 'False'
|
||||
END AS 'Fall 1.1'
|
||||
""",
|
||||
{"parent": parent_partnum, "child": child_id}).fetchone()
|
||||
{"parent": self.parent_code, "child": child_code}).fetchone()
|
||||
|
||||
return True if res["Fall 1.1"] == 'True' else False
|
||||
|
||||
def occurences_at_same_pos_in_both_boms(parent_partnum: str, child_id: str) -> bool:
|
||||
def _check_same_pos(self, child_code: str) -> bool:
|
||||
"""
|
||||
Prüfung Fall 1.2
|
||||
"""
|
||||
res = con.execute("""
|
||||
res = self.db.con.execute("""
|
||||
SELECT CASE
|
||||
WHEN NOT EXISTS (
|
||||
-- Prüfung, ob in der neuen Stückliste an denselben Positionen wie in der alten Stückliste Längenvarianten zur angegebenen Teilenr. vorliegen
|
||||
@@ -167,56 +450,15 @@ def occurences_at_same_pos_in_both_boms(parent_partnum: str, child_id: str) -> b
|
||||
ELSE 'False'
|
||||
END AS 'Fall 1.2'
|
||||
""",
|
||||
{"parent": parent_partnum, "child": child_id}).fetchone()
|
||||
{"parent": self.parent_code, "child": child_code}).fetchone()
|
||||
return True if res["Fall 1.2"] == 'True' else False
|
||||
|
||||
def get_data_for_new_pos(parent_partnum: str, pos: str) -> list[dict[str, str]]:
|
||||
"""
|
||||
Fall 1.3 (Neuanlage von Positionen)
|
||||
"""
|
||||
res = con.execute("""
|
||||
SELECT RD.Positionsnummer AS POSNR,
|
||||
RD.ChildCode AS KOMPTEILENR,
|
||||
RD.ParentCode AS PARENT,
|
||||
CASE -- Abw. Anz. = NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV
|
||||
WHEN RD.Uebergabelaenge = 0 THEN NULL
|
||||
ELSE RD.Menge
|
||||
END AS ABW_ANZAHL,
|
||||
CASE -- Mengenwert berechnen, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV
|
||||
WHEN RD.Uebergabelaenge = 0 THEN RD.Menge
|
||||
ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT,0)) / 1000
|
||||
END AS MENGE,
|
||||
CASE -- LAENGE = NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV
|
||||
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
|
||||
|
||||
-- Da die Positionen neu angelegt werden, werden die Defaultwerte verwendet, die zum Teil aus dem Teilestamm (PARTS) kommen.
|
||||
LEFT JOIN PARTS P
|
||||
ON RD.ChildCode = P.TEILENR
|
||||
|
||||
WHERE C.KOMPTEILENR IS NULL
|
||||
AND RD.ParentCode = :parent
|
||||
AND RD.Positionsnummer = :posnr
|
||||
""",
|
||||
{"parent": parent_partnum, "posnr": pos}).fetchone()
|
||||
return res
|
||||
|
||||
def get_update_data_for_diff_pos(parent_partnum: str, pos: str) -> list[dict[str, str]]:
|
||||
def _get_update_data_for_diff_pos(self, pos_nr: str) -> list[dict[str]]:
|
||||
"""
|
||||
Ermittelt die Daten, die an die neue Position verschoben werden sollen.
|
||||
(Für Fall 1.1 und Fall 1.3)
|
||||
"""
|
||||
res = con.execute("""
|
||||
res = self.db.con.execute("""
|
||||
SELECT C.ID,
|
||||
C.DISPO_AKTIV_JN,
|
||||
C.ZUSCHNITT,
|
||||
@@ -258,15 +500,15 @@ def get_update_data_for_diff_pos(parent_partnum: str, pos: str) -> list[dict[str
|
||||
WHERE RD.ParentCode = :parent
|
||||
AND RD.Positionsnummer = :posnr
|
||||
""",
|
||||
{"parent": parent_partnum, "posnr": pos}).fetchone()
|
||||
{"parent": self.parent_code, "posnr": pos_nr}).fetchone()
|
||||
return res
|
||||
|
||||
def get_update_data_for_same_pos(parent_partnum: str, pos: str) -> list[dict[str, str]]:
|
||||
def _get_update_data_for_same_pos(self, pos_nr: str) -> list[dict[str]]:
|
||||
"""
|
||||
Ermittelt die Daten, die an dieselbe Position verschoben werden sollen
|
||||
(Für Fall 1.2)
|
||||
"""
|
||||
res = con.execute("""
|
||||
res = self.db.con.execute("""
|
||||
SELECT C.ID,
|
||||
C.ZUSCHNITT,
|
||||
C.DISPO_AKTIV_JN,
|
||||
@@ -308,202 +550,146 @@ def get_update_data_for_same_pos(parent_partnum: str, pos: str) -> list[dict[str
|
||||
WHERE RD.ParentCode = :parent
|
||||
AND RD.Positionsnummer = :posnr
|
||||
""",
|
||||
{"parent": parent_partnum, "posnr": pos}).fetchone()
|
||||
{"parent": self.parent_code, "posnr": pos_nr}).fetchone()
|
||||
return res
|
||||
|
||||
def create_new_bom(parent_partnum:str) -> None:
|
||||
def _get_ids_to_del(self, ignored_ids: list = []) -> list[dict[str]]:
|
||||
"""
|
||||
Die zu löschenden ID's ergeben sich, indem man die ID's der Positionen aus der alten Stückliste betrachtet
|
||||
und die ID's aus der neuen Stückliste abzieht, die verschoben werden. Positionen, die in der neuen Stückliste
|
||||
neu dazukommen sind, wirken sich nicht auf die zu löschenden ID's aus!
|
||||
"""
|
||||
old_bom_ids = self.old_bom.get_child_ids(self.parent_code)
|
||||
ids_to_delete = [idx for idx in old_bom_ids if idx["ID"] not in ignored_ids]
|
||||
return ids_to_delete
|
||||
|
||||
new_bom_data = get_new_bom_data(parent_partnum)
|
||||
def _get_data_for_insert(self, posnr: str) -> list[dict[str]]:
|
||||
"""
|
||||
Fall 1.3 (Neuanlage von Positionen)
|
||||
"""
|
||||
res = self.db.con.execute("""
|
||||
SELECT RD.Positionsnummer AS POSNR,
|
||||
RD.ChildCode AS KOMPTEILENR,
|
||||
RD.ParentCode AS PARENT,
|
||||
NULL AS ZUSCHNITT,
|
||||
P.DISPO_AKTIV_JN,
|
||||
P.RUESTCODE,
|
||||
|
||||
# Ermittlung, welche Teilenummern unter Fall 1.1 , 1.2 oder 1.3 fallen
|
||||
# Nicht im folgenden Loop, um mehrfache Fallprüfung 1.1 und 1.2 zu vermeiden (wenn längenabhängige Teilenummern mehrfach in Stückliste vorliegen)
|
||||
partnum_cases = {}
|
||||
for result_line in new_bom_data:
|
||||
child_partnum = result_line["ChildCode"]
|
||||
if child_partnum not in partnum_cases:
|
||||
if identical_occurences_in_both_boms(parent_partnum, child_partnum):
|
||||
partnum_cases[child_partnum] = 1.1
|
||||
elif occurences_at_same_pos_in_both_boms(parent_partnum, child_partnum):
|
||||
partnum_cases[child_partnum] = 1.2
|
||||
else:
|
||||
partnum_cases[child_partnum] = 1.3
|
||||
-- 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,
|
||||
|
||||
# Ermittlung unter welchen Fall der Logik, die jeweilige Positionsnummer der neuen Stückliste fällt
|
||||
pos_to_update = [] # Sammelt Positionsnummern (inkl. Daten), die upgedated werden sollen
|
||||
pos_to_insert = [] # Sammelt Positionsnummern (inkl. Daten), die eingefügt werden sollen
|
||||
pos_to_del = [] # Sammelt Positionsnummern, die gelöscht werden sollen
|
||||
moved_partnums = [] # Sammelt Positionsnummern, die verschoben werden. Dient später zur Ermittlung der zu löschenden Positionen
|
||||
-- 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
|
||||
|
||||
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
|
||||
|
||||
-- Da es sich um neue Positionen handelt, werden Defaultwerte verwendet, die z.T. aus dem Teilestamm (PARTS) kommen
|
||||
LEFT JOIN PARTS P
|
||||
ON RD.ChildCode = P.TEILENR
|
||||
|
||||
WHERE C.KOMPTEILENR IS NULL
|
||||
AND RD.ParentCode = :parent
|
||||
AND RD.Positionsnummer = :posnr
|
||||
""",
|
||||
{"parent": self.parent_code, "posnr": posnr}).fetchone()
|
||||
return res
|
||||
|
||||
def write_new_bom(self) -> None:
|
||||
"""
|
||||
Diese Methode binhaltet die Hauptlogik:
|
||||
Hier wird festgelegt, welche Positionsnummern eingefügt, gelöscht oder verschoben werden müssen
|
||||
Am Ende werden die daraus erforderlichen Datenbankaktionen (insert / update / delete) initiiert,
|
||||
um die neue, fertig verarbeite Stückliste in COMP zu überführen.
|
||||
"""
|
||||
update_data_comp = []
|
||||
insert_data_comp = []
|
||||
update_data_regel = []
|
||||
update_data_bedingung = []
|
||||
ids_to_delete = []
|
||||
ids_not_to_delete = []
|
||||
childcodes_case_11 = set() # vermeidet erneute Prüfung einer Teilenummer durch _check_identical_code_len_combo
|
||||
childcodes_case_12 = set() # vermeidet erneute Prüfung einer Teilenummer durch _check_same_pos()
|
||||
|
||||
new_bom_data = self.new_bom.get_all_bom_data()
|
||||
|
||||
for pos_data in new_bom_data:
|
||||
child_partnum = pos_data["ChildCode"]
|
||||
child_pos = pos_data["Positionsnummer"]
|
||||
|
||||
# Fall 1.1 (Kann nicht mit 1. Fall in Fall 1.3.1 (siehe unten) fusioniert werden, weil sonst z.B. bei Beispiel 3 fälschlicherweise diese Logik greifen würde)
|
||||
if partnum_cases[child_partnum] == 1.1:
|
||||
data = get_update_data_for_diff_pos(parent_partnum, child_pos)
|
||||
pos_to_update.append(data)
|
||||
moved_partnums.append(data["ID"])
|
||||
child_code: str = pos_data["ChildCode"]
|
||||
posnr: int = pos_data["Positionsnummer"]
|
||||
|
||||
# Fall 1.2
|
||||
elif partnum_cases[child_partnum] == 1.2:
|
||||
data = get_update_data_for_same_pos(parent_partnum, child_pos)
|
||||
pos_to_update.append(data)
|
||||
moved_partnums.append(data["ID"])
|
||||
# Prüfung Fall 1.1
|
||||
if child_code in childcodes_case_11 or self._check_identical_code_len_combo(child_code) is True:
|
||||
logic_case = 1.1
|
||||
childcodes_case_11.add(child_code)
|
||||
data = self._get_update_data_for_diff_pos(posnr)
|
||||
|
||||
# Fall 1.3
|
||||
elif partnum_cases[child_partnum] == 1.3:
|
||||
# Prüfung Fall 1.2
|
||||
elif child_code in childcodes_case_12 or self._check_same_pos(child_code) is True:
|
||||
logic_case = 1.2
|
||||
childcodes_case_12.add(child_code)
|
||||
data = self._get_update_data_for_same_pos(posnr)
|
||||
|
||||
# Fall 1.3.1 - Verschiebung von Positionen
|
||||
pos_to_move = get_positions_to_move(parent_partnum)
|
||||
if child_pos in pos_to_move:
|
||||
data = get_update_data_for_diff_pos(parent_partnum, child_pos)
|
||||
pos_to_update.append(data)
|
||||
moved_partnums.append(data["ID"])
|
||||
|
||||
# Fall 1.3.2 - Neuanlage von Positionen
|
||||
# Prüfungen Fall 1.3
|
||||
else:
|
||||
data = get_data_for_new_pos(parent_partnum, child_pos)
|
||||
pos_to_insert.append(data)
|
||||
|
||||
# Prüfung Fall 1.31 - Verschiebung bestehender Positionen
|
||||
posnrs_to_move = self._get_positions_to_move_old_logic()
|
||||
if posnr in posnrs_to_move:
|
||||
logic_case = 1.31
|
||||
data = self._get_update_data_for_diff_pos(posnr)
|
||||
|
||||
# Prüfung Fall 1.32 - Einfügen neuer Positionen
|
||||
else:
|
||||
raise NotImplementedError("Fall ist nicht vorgesehen (dürfte laut spezifizierter Logik nicht auftreten). Prüfung erforderlich!")
|
||||
logic_case = 1.32
|
||||
data = self._get_data_for_insert(posnr)
|
||||
|
||||
# Positionen aus der alten Stückliste, die nicht verschoben werden müssen gelöscht werden.
|
||||
# Positionen, die in der neuen Stückliste neu eingefügt werden wirken sich nicht auf die Positionen der alten Stückliste aus und bleiben
|
||||
# deshalb bei der Bestimmung der zu löschenden Positionen unberücksichtigt.
|
||||
old_bom_ids = get_old_bom_ids(parent_partnum)
|
||||
for idx in old_bom_ids:
|
||||
if idx["ID"] not in moved_partnums:
|
||||
pos_to_del.append(idx)
|
||||
# Teilenummern der Stückliste, die verschoben werden (Fal 1.1 / 1.2 / 1.3.1):
|
||||
# - werden nie gelöscht (konkret: deren ID's in COMP), weil IMMER eine Verschiebung der Positionsnummer erfolgt
|
||||
if logic_case in (1.1, 1.2, 1.31):
|
||||
ids_not_to_delete.append(data["ID"])
|
||||
update_data_comp.append(data)
|
||||
|
||||
# Datenbankaktionen durchführen, um die neue Stückliste zu schreiben.
|
||||
if pos_to_del:
|
||||
delete_pos(pos_to_del)
|
||||
if pos_to_insert:
|
||||
insert_pos(pos_to_insert)
|
||||
if pos_to_update:
|
||||
update_pos(pos_to_update)
|
||||
|
||||
def insert_pos(new_data: dict[str, str]) -> None:
|
||||
"""
|
||||
Fügt die Positionen in die Stücklist ein, bei denen keine Verschiebung von der alten Stückliste erfolgt ist.
|
||||
Diese Positionen werden stets ohne TOS-Regeln oder -Bedingungen angelegt.
|
||||
"""
|
||||
con.executemany("""INSERT INTO COMP (PARENT, POSNR, KOMPTEILENR, DISPO_AKTIV_JN, MENGE, ABW_ANZAHL, LAENGE, ZUSCHNITT, RUESTCODE)
|
||||
VALUES (:PARENT, :POSNR, :KOMPTEILENR, :DISPO_AKTIV_JN, :MENGE, :ABW_ANZAHL, :LAENGE, :ZUSCHNITT, :RUESTCODE)""",
|
||||
new_data)
|
||||
|
||||
def delete_pos(ids: list[dict[str, str]]) -> None:
|
||||
"""
|
||||
- Entfernt die Positionen aus der Stückliste (COMP), die nicht verschoben werden und nicht mehr in der alten Stückliste vorliegen.
|
||||
- Entfernt die TOS-Regeln aus STKL_REGEL und TOS-Bedingungen aus VAR_TEIL_BEDINGUNG_2 zu den oben genannten Positionen.
|
||||
"""
|
||||
con.executemany("DELETE FROM COMP WHERE ID = :ID", ids)
|
||||
con.executemany("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE COMP_ID = :ID", ids)
|
||||
con.executemany("DELETE FROM STKL_REGEL WHERE COMP_ID = :ID", ids)
|
||||
|
||||
def update_pos(new_data: list[dict[str, str]]) -> None:
|
||||
"""
|
||||
- Verschiebt Positionsdaten von einer Position in der alten Stückliste zu einer Position in der neuen Stückliste
|
||||
- Aktualisiert TOS-Regeln in STKL_REGEL und
|
||||
"""
|
||||
con.executemany("""
|
||||
UPDATE COMP
|
||||
SET POSNR = :POSNR,
|
||||
ABW_ANZAHL = :ABW_ANZAHL,
|
||||
MENGE = :MENGE,
|
||||
LAENGE = :LAENGE,
|
||||
ZUSCHNITT = :ZUSCHNITT,
|
||||
DISPO_AKTIV_JN = :DISPO_AKTIV_JN,
|
||||
RUESTCODE = :RUESTCODE
|
||||
WHERE ID = :ID
|
||||
""", new_data)
|
||||
|
||||
update_bedingungen = [pos_data for pos_data in new_data if pos_data["BEDINGUNG"] is not None]
|
||||
update_regeln = [pos_data for pos_data in new_data if pos_data["REGEL"] is not None]
|
||||
|
||||
if update_bedingungen:
|
||||
con.executemany("UPDATE VAR_TEIL_BEDINGUNG_2 SET BEDINGUNG = :BEDINGUNG WHERE COMP_ID = :ID", update_bedingungen)
|
||||
if update_regeln:
|
||||
con.executemany("UPDATE STKL_REGEL SET REGEL = :REGEL WHERE COMP_ID = :ID", update_regeln)
|
||||
|
||||
def print_new_bom(parent_partnum: str) -> None:
|
||||
res = con.execute("""
|
||||
SELECT C.POSNR AS Pos,
|
||||
C.KOMPTEILENR AS 'Art.-Nr.',
|
||||
C.MENGE As Menge,
|
||||
C.ABW_ANZAHL AS 'Abw. Anzahl',
|
||||
C.LAENGE AS Länge,
|
||||
C.ZUSCHNITT AS Zuschnitt,
|
||||
B.BEDINGUNG AS Bedingung,
|
||||
R.REGEL AS Regel,
|
||||
BSP.Beispielnr AS Beispiel
|
||||
|
||||
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
|
||||
|
||||
LEFT JOIN Beispiele BSP
|
||||
ON C.PARENT = BSP.Parent_ID
|
||||
|
||||
WHERE C.PARENT = :parent
|
||||
ORDER BY POSNR
|
||||
""",
|
||||
{"parent": parent_partnum}).fetchall()
|
||||
|
||||
for line in res:
|
||||
print(line)
|
||||
|
||||
def write_bom_as_csv(parent_partnum: str) -> None:
|
||||
out_path = f".data/BOM_{parent_partnum}_new.csv"
|
||||
|
||||
res: list[dict[str, str]]
|
||||
res = con.execute("""
|
||||
SELECT C.POSNR AS Pos,
|
||||
C.KOMPTEILENR AS 'Art.-Nr.',
|
||||
C.MENGE As Menge,
|
||||
C.ABW_ANZAHL AS 'Abw. Anzahl',
|
||||
C.LAENGE AS Länge,
|
||||
C.ZUSCHNITT AS Zuschnitt,
|
||||
B.BEDINGUNG AS Bedingung,
|
||||
R.REGEL AS Regel,
|
||||
BSP.Beispielnr AS Beispiel
|
||||
|
||||
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
|
||||
|
||||
LEFT JOIN Beispiele BSP
|
||||
ON C.PARENT = BSP.Parent_ID
|
||||
|
||||
WHERE C.PARENT = :parent
|
||||
ORDER BY POSNR
|
||||
""",
|
||||
{"parent": args["partnumber"]}).fetchall()
|
||||
|
||||
with open("test.csv", "w", encoding="utf-8", newline="") as csv_file:
|
||||
writer = csv.DictWriter(csv_file, delimiter=";", fieldnames=res[0].keys())
|
||||
writer.writeheader()
|
||||
writer.writerows(res)
|
||||
if logic_case == 1.32:
|
||||
insert_data_comp.append(data)
|
||||
|
||||
|
||||
# Zu löschende Positionen ermitteln
|
||||
ids_to_delete = self._get_ids_to_del(ignored_ids=ids_not_to_delete)
|
||||
|
||||
# Positionen löschen (falls erforderlich)
|
||||
if ids_to_delete:
|
||||
self.new_bom.delete(ids_to_delete)
|
||||
self.tos_bedingungen.delete(ids_to_delete)
|
||||
self.tos_regeln.delete(ids_to_delete)
|
||||
|
||||
# Positionen einfügen (falls erforderlich). Neue Positionen haben nie TOS-Regeln oder -Bedingungen
|
||||
if insert_data_comp:
|
||||
self.new_bom.insert(insert_data_comp)
|
||||
|
||||
# Positionen sowie deren TOS-Regeln und -Bedingungen updaten (falls erforderlich)
|
||||
if update_data_comp:
|
||||
update_data_regel = self.extract_col_data(update_data_comp, "ID", "REGEL")
|
||||
update_data_bedingung = self.extract_col_data(update_data_comp, "ID", "BEDINGUNG")
|
||||
|
||||
self.new_bom.update(update_data_comp)
|
||||
self.tos_regeln.update(update_data_regel)
|
||||
self.tos_bedingungen.update(update_data_bedingung)
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("-cd", "--csv_dir", help="Verzeichnis, in der CSV-Datei mit den neuen Stücklisteninformationen liegt", default=r".\data")
|
||||
parser.add_argument("-sd", "--sql_dir", help="Verzeichnis, in der SQL-Datei mit den Informationen zur Teilreplizierung der SIVAS-Datenbank liegt", default=r".\data")
|
||||
parser.add_argument("-n", "--partnumber", help="Teilenummer der Baugruppe", type=str, default="000000382")
|
||||
args: dict = vars(parser.parse_args())
|
||||
|
||||
# Vorerst wird echte bei jedem Lauf neue db-Datei erstellt, da es mit Datenbank im RAM (:memory:) teilweise zu fehlerhaften Abfragen kam
|
||||
if "sivas_replika.db" in os.listdir("."):
|
||||
os.remove("./sivas_replika.db")
|
||||
|
||||
with closing(sqlite3.connect("./sivas_replika.db", autocommit=True)) as con:
|
||||
con.row_factory = dict_factory
|
||||
main(args["partnumber"], args["sql_dir"], args["csv_dir"])
|
||||
main()
|
||||
Reference in New Issue
Block a user