Das deutsche Python-Forum

Code: Alles auswählen

import sys
import tkinter as tk
from datetime import datetime
from functools import partial
from pathlib import Path
from queue import Empty, Queue
from threading import Event, Thread
from tkinter import messagebox, ttk
from tkinter.filedialog import askdirectory

import xlwings as xw
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Alignment

TABLE_NAME = "Messreihe"
WORKBOOK_WITH_READ_DATA = "Ausgelesene Werte.xlsx"
WORKBOOK_WITH_READ_CALCULATED_DATA = "Ausgelesene und berechnete Werte.xlsx"
CALCULATION_WORKBOOK = "Ausl.-progr"
INFO_FILE = "Nicht gelesene Dateien.txt"
ALLOWED_SUFFIXES = [".xlsx", ".xlsm"]
# WORKING_PATH = Path(sys.executable).parent
WORKING_PATH = Path(r"C:\Users\straub\Documents")

# For edit use following syntax:
# custom description: {"protocol": cell-name of protocol, "calculation": cell-name of calculation}
DESCRIPTION_TO_CELL = {
    "Auftragsnummer": {"protocol": "E7", "calculation": "G6"},
    "Code": {"protocol": "X7", "calculation": "G4"},
    "Austritttsdruck 1": {"protocol": "K50", "calculation": None},
    "Wirkleistungsaufnahmen": {"protocol": "K68", "calculation": None,},
    "Summe Normvolumenstrom [Nm³/h]": {"protocol": "L119", "calculation": None},
    "Neu berechneter Normvolumenstrom [Nm³/h]": {"protocol": None, "calculation": None},
    "Differenz Normvolumenstrom [Nm³/h]": {"protocol": None, "calculation": None},
    "Differenz Normvolumenstrom [%]": {"protocol": None, "calculation": None},
}


class App(tk.Frame):
    def __init__(self, master):
        tk.Frame.__init__(self, master)
        ttk.Label(self, text="Ausgewählter Ordner:").grid(
            row=0, column=0, sticky=tk.W, ipadx=30
        )
        self.user_select = tk.StringVar()
        self.user_select.set("<Kein Oderne ausgewählt>")
        self.selected_folder = ttk.Label(self, textvariable=self.user_select)
        self.selected_folder.grid(row=1, column=0, sticky=tk.W)
        ttk.Button(self, text="Ordner auswählen", command=self.open_dialog).grid(
            row=2, column=0, sticky=tk.W
        )
        ttk.Button(
            self, text="Auslesen", command=partial(self.transfer_gui_data, False)
        ).grid(row=5, column=1, sticky=tk.E)
        ttk.Button(
            self,
            text="Auslesen & Berechnen",
            command=partial(self.transfer_gui_data, True),
        ).grid(row=5, column=2, sticky=tk.E)
        ttk.Button(self, text="Beenden", command=self.stop_program).grid(
            row=6, column=2, sticky=tk.E
        )
        self.process_status = tk.StringVar()
        ttk.Label(self, textvariable=self.process_status, font=("Arial", 15)).grid(
            row=2, column=1, columnspan=2, sticky=tk.E
        )

        self.cancel = Event()
        self.process = Queue()
        self.calculation = None
        self.folder_path = None

    def on_progress(self, percentage):
        self.process.put(percentage)

    def open_dialog(self):
        self.folder_path = Path(askdirectory(mustexist=True))
        self.user_select.set(self.folder_path.name)
        self.process_status.set("Bereit zum lesen 🤓")

    @staticmethod
    def show_info_file_message():
        messagebox.showinfo(
            "Info Datei",
            "Nicht alle gefundenen Dateien konnten gelesen werden, bitte Info-Datei beachten.",
        )

    def stop_program(self):
        self.cancel.set()
        self.quit()

    def update_process(self):
        try:
            percent = self.process.get(block=False)
        except Empty:
            pass
        else:
            if percent == 100:
                self.process_status.set("Fertig! 😎")
                return
            try:
                int(percent)
                self.process_status.set(f"Time for Coffee ☕: {percent:.0f} %")
            except ValueError:
                self.process_status.set("Auf Excel warten 😴 ")
        self.after(10, self.update_process)

    def transfer_gui_data(self, calculation_state):
        Thread(
            target=control_program_sequence,
            args=[
                self.folder_path,
                self.cancel,
                self.show_info_file_message,
                calculation_state,
                self.on_progress,
            ],
        ).start()
        self.update_process()


def calculate_actually_volume(workbook, cell_content):
    for index, cell in enumerate(DESCRIPTION_TO_CELL):
        if DESCRIPTION_TO_CELL[cell]["calculation"] is not None:
            workbook.sheets["Berechnungsblatt"].range(
                DESCRIPTION_TO_CELL[cell]["calculation"]
            ).value = cell_content[index]
    return workbook.sheets["Berechnungsblatt"].range("O33").value


def calculate_volume_difference(cell_contents):
    try:
        old_volume = float(cell_contents[-2])
        new_volume = float(cell_contents[-1])
        difference = abs(old_volume - new_volume)
        return difference, 100 * difference / old_volume
    except TypeError:
        return "-" * 2


def collect_files(folder_path):
    return [
        file
        for file in folder_path.iterdir()
        if file.is_file()
        and file.suffix in ALLOWED_SUFFIXES
        and file.name.split("_")[-1] != WORKBOOK_WITH_READ_DATA
    ]


def control_program_sequence(
    folder_path,
    cancel,
    show_info_file_message,
    calculation,
    on_progress=lambda percentage: None,
):
    info_file = False
    timestamp = datetime.now().strftime("%Y%m%d_%H%M")
    workbook, output_file = open_output_file(timestamp, calculation)
    worksheet = workbook.active
    write_column_description(worksheet)
    if calculation:
        on_progress("-")
        calculation_workbook = search_calculation_file()
        with xw.App(visible=False, add_book=False) as excel_app:
            excel_app.display_alerts = False
            excel_calculation = excel_app.books.open(
                calculation_workbook, update_links=False
            )
            control_read_write_data(
                folder_path,
                cancel,
                show_info_file_message,
                worksheet,
                workbook,
                timestamp,
                output_file,
                info_file,
                on_progress,
                excel_calculation,
            )
    else:
        control_read_write_data(
            folder_path,
            cancel,
            show_info_file_message,
            worksheet,
            workbook,
            timestamp,
            output_file,
            info_file,
            on_progress,
        )


def control_read_write_data(
    folder_path,
    cancel,
    show_info_file_message,
    worksheet,
    workbook,
    timestamp,
    output_file,
    info_file,
    on_progress=lambda percentage: None,
    excel_calculation=None,
):
    files = collect_files(folder_path)
    for number, filename in enumerate(files, 1):
        if cancel.is_set():
            break
        on_progress(number / len(files) * 100)
        cell_contents = read_excel_file(filename)
        if cell_contents:
            if excel_calculation is not None:
                cell_contents.append(
                    calculate_actually_volume(excel_calculation, cell_contents)
                )
                cell_contents.extend(iter(calculate_volume_difference(cell_contents)))
            write_into_output_file(worksheet, cell_contents)
        else:
            write_info_file(timestamp, filename)
            info_file = True
    workbook.save(output_file)
    if info_file:
        show_info_file_message()


def search_calculation_file():
    for file in WORKING_PATH.iterdir():
        if CALCULATION_WORKBOOK in file.name:
            return file


def open_output_file(timestamp, calculation):
    if calculation:
        output_file = WORKING_PATH / f"{timestamp}_{WORKBOOK_WITH_READ_CALCULATED_DATA}"
    else:
        output_file = WORKING_PATH / f"{timestamp}_{WORKBOOK_WITH_READ_DATA}"
    workbook = load_workbook(output_file) if output_file.exists() else Workbook()
    return workbook, output_file


def read_excel_file(filename):
    try:
        worksheet = load_workbook(filename, data_only=True)[TABLE_NAME]
    except KeyError:
        return []
    return [
        worksheet[DESCRIPTION_TO_CELL[description]["protocol"]].value
        for description in DESCRIPTION_TO_CELL
        if DESCRIPTION_TO_CELL[description]["protocol"] is not None
    ]


def write_column_description(worksheet):
    for column, description in enumerate(DESCRIPTION_TO_CELL, 1):
        cell = worksheet.cell(
            row=1,
            column=column,
        )
        cell.alignment = Alignment(textRotation=90)
        cell.value = description


def write_info_file(timestamp, file):
    with (WORKING_PATH / f"{timestamp}_{INFO_FILE}").open(
        "a", encoding="UTF-8"
    ) as info_file:
        info_file.write(
            f'-> Datei "{file}" konnte nicht nach gegebenen Kriterien ausgelesen werden.\n'
        )


def write_into_output_file(worksheet, cell_contents):
    worksheet.append(cell_contents)


def main():
    root = tk.Tk()
    root.title("Messdateien auslesen")
    app = App(root)
    app.pack()
    app.mainloop()


if __name__ == "__main__":
    main()

Code: Alles auswählen

def calculate_total_volume(workbook, cell_content):
    berechnungsblatt = workbook.sheets["Berechnungsblatt"]
    for value, cell_ranges in zip(cell_content, DESCRIPTION_TO_CELL.values()):
        if "calculation" in cell_ranges:
            berechnungsblatt.range(
                cell_ranges["calculation"]
            ).value = value
    return berechnungsblatt.range("O33").value

Code: Alles auswählen

"Differenz Normvolumenstrom [%]": {"protocol": None, "calculation": None}