Dieser Artikel basiert auf dem Thema des Qiita Summer Festival 2020 "Wenn Sie eine △△ (App) nur mit 〇〇 (Sprache) erstellen möchten" Der Inhalt stimmt damit überein.
Am 5. Juli 2020 wurde meine Arbeit "VMD-Größe ver5.00 (exe code Code)" Wurde veröffentlicht. Das Thema dieses Tools ist "Regenerieren von VMD (MMD-Bewegungsdaten) mit einem geeigneten Kopf und Körper für ein bestimmtes Modell". Für die meisten von Ihnen, die diesen Artikel lesen, klingt dies so. Es ist eine komplette Hobby-App.
** Nga! !! !! ** ** **
Die meisten Leute, die MMD (MikuMikuDance) mögen, sind ganz normale Leute, die nichts mit Python oder Programmen zu tun haben. Wenn jemand mit "Schlange ?!" Antwortet, finde ich das ziemlich wertvoll. Wie können solche Leute ihre eigenen Apps nutzen? Dieser Artikel ist an die Spitze solcher Qualen, Versuche und Irrtümer und vieler Probleme geraten. Es ist eine Zusammenfassung meiner eigenen Antworten. Übrigens wurde "VMD-Dimensionierung" in dem Maße verwendet, in dem die kumulierte Summe DL8500 überschreitet. (Ungefähr 9600 DL in Kombination mit der 32-Bit-Version)
Python + pyinstaller = exe
ist nicht so selten, aber es erfordert etwas Einfallsreichtum, um es auf ein Niveau zu bringen, das dem tatsächlichen Betrieb standhält.
<"Ist es nicht in Ordnung, es in C zu schaffen?" ** Ich mag Python. ** (Weil ich C nicht kenne ...)
Lassen Sie uns zunächst die Entwicklungsumgebung vorbereiten.
<"Es ist kein Grund, maschinell zu lernen. Ist es nicht in Ordnung, es roh zu lassen?" **Auf keinen Fall! !! ** ** **
Ein häufiges Problem in Pyinstaller-Artikeln ist, dass "zusätzliche Bibliotheken enthalten sind und die exe-Datei groß wird". Es ist üblich, während der Entwicklung neue Bibliotheken auszuprobieren. Wenn Sie jedoch eine Exe so wie sie ist erstellen, besteht eine hohe Wahrscheinlichkeit, dass unnötige Bibliotheken in die Exe aufgenommen werden. ** Lassen Sie uns die Entwicklungsumgebung und die Release-Umgebung genau trennen. ** ** **
Laden Sie das Installationsprogramm von [Anaconda Official] herunter (https://www.anaconda.com/products/individual). Wenn es von nun an hergestellt wird, ist es besser, 3er zu verwenden.
Befolgen Sie nach DL die Installationsschritte.
Lassen Sie uns zunächst eine virtuelle Umgebung für die Entwicklung erstellen.
conda create -n pytest_env pip python=3.7
Sobald Sie eine Entwicklungsumgebung haben, lassen Sie uns "aktivieren". Lassen Sie uns übrigens auch ein Verwaltungsverzeichnis für den Quellcode erstellen.
Erstellen Sie auf ähnliche Weise eine Release-Umgebung.
conda create -n pytest_release pip python=3.7
Sobald Sie das Verwaltungsverzeichnis haben, wechseln Sie dorthin und installieren Sie die erforderlichen Bibliotheken. Hier ist ein Trick.
** pyinstaller
wird nur in der Release-Umgebung installiert **
Indem Sie "pyinstaller" nur in der Release-Umgebung installieren, können Sie Fehler vermeiden, die Sie versehentlich in der Entwicklungsumgebung veröffentlichen. Da es eine große Sache ist, lassen Sie uns numpy einführen.
Installationsbefehl für die Entwicklungsumgebung
pip install numpy wxPython
Installationsbefehl für die Release-Umgebung
pip install numpy wxPython pypiwin32 pyinstaller
pypiwin32
scheint die Bibliothek zu sein, die zum Ausführen von pyinstaller
unter Windows benötigt wird.
Die GUI ist einfach mit WxFormBuilder zu erstellen. Es gibt jedoch einige Dinge, die für die automatische Namenskonvention schwer zu verstehen sind, dass Teile nicht wiederverwendet werden können und dass einige Dinge für die Erstellung einer tatsächlichen Operationsanwendung etwas unbefriedigend sind. Deshalb gebe ich sie aus, wenn sie in einer bestimmten Form vorliegt, und danach muss ich sie selbst ausführen. Ich empfehle es.
Referenz: GUI (WxFormBuilder) in Python (mm_sys) https://qiita.com/mm_sys/items/716cb159ea8c9e634300
Danach werden wir in einem Reverse-Pull-Format fortfahren. Bitte lesen Sie den Abschnitt, an dem Sie interessiert sind.
Ich denke, die meisten Leute, die sich für diesen Artikel interessieren, interessieren sich dafür. Ich würde gerne wissen, ob es eine richtige Antwort gibt. Also habe ich verschiedene Dinge übersprungen und an den Anfang gebracht.
--Starten Sie einen Logik-Thread, der lange ausgeführt wird, während der GUI-Thread unverändert bleibt --Multiprozess kann auch im Logik-Thread ausgeführt werden --Logic-Threads werden dem Prozess nicht hinzugefügt
Unten finden Sie den Code, der die oben genannten Anforderungen erfüllt.
executor.py
# -*- coding: utf-8 -*-
#
import wx
import sys
import argparse
import numpy as np
import multiprocessing
from pathlib import Path
from form.MainFrame import MainFrame
from utils.MLogger import MLogger
VERSION_NAME = "ver1.00"
#Keine Exponenten-Notation, wird weggelassen, wenn die Anzahl der effektiven Dezimalstellen 6, 30 Zeichen, 200 Zeichen pro Zeile überschreitet
np.set_printoptions(suppress=True, precision=6, threshold=30, linewidth=200)
#Windows-Multiprozessmaßnahmen
multiprocessing.freeze_support()
if __name__ == '__main__':
#Argumentinterpretation
parser = argparse.ArgumentParser()
parser.add_argument("--verbose", default=20, type=int)
args = parser.parse_args()
#Logger-Initialisierung
MLogger.initialize(level=args.verbose, is_file=False)
#GUI-Start
app = wx.App(False)
frame = MainFrame(None, VERSION_NAME, args.verbose)
frame.Show(True)
app.MainLoop()
Zuerst der Aufrufer executor.py
.
Starten Sie die GUI von hier aus.
MainFrame.py
# -*- coding: utf-8 -*-
#
from time import sleep
from worker.LongLogicWorker import LongLogicWorker
from form.ConsoleCtrl import ConsoleCtrl
from utils.MLogger import MLogger
import os
import sys
import wx
import wx.lib.newevent
logger = MLogger(__name__)
TIMER_ID = wx.NewId()
(LongThreadEvent, EVT_LONG_THREAD) = wx.lib.newevent.NewEvent()
#Haupt-GUI
class MainFrame(wx.Frame):
def __init__(self, parent, version_name: str, logging_level: int):
self.version_name = version_name
self.logging_level = logging_level
self.elapsed_time = 0
self.worker = None
#Initialisieren
wx.Frame.__init__(self, parent, id=wx.ID_ANY, title=u"c01 Long Logic {0}".format(self.version_name), \
pos=wx.DefaultPosition, size=wx.Size(600, 650), style=wx.DEFAULT_FRAME_STYLE)
self.sizer = wx.BoxSizer(wx.VERTICAL)
#Verarbeitungszeit
self.loop_cnt_ctrl = wx.SpinCtrl(self, id=wx.ID_ANY, size=wx.Size(100, -1), value="2", min=1, max=999, initial=2)
self.loop_cnt_ctrl.SetToolTip(u"Verarbeitungszeit")
self.sizer.Add(self.loop_cnt_ctrl, 0, wx.ALL, 5)
#Kontrollkästchen für Parallelverarbeitung
self.multi_process_ctrl = wx.CheckBox(self, id=wx.ID_ANY, label="Wenn Sie die Parallelverarbeitung ausführen möchten, überprüfen Sie diese bitte.")
self.sizer.Add(self.multi_process_ctrl, 0, wx.ALL, 5)
#Button Sizer
self.btn_sizer = wx.BoxSizer(wx.HORIZONTAL)
#Schaltfläche "Ausführen"
self.exec_btn_ctrl = wx.Button(self, wx.ID_ANY, u"Starten Sie die lange Logikverarbeitung", wx.DefaultPosition, wx.Size(200, 50), 0)
#Mit Mausklick binden Ereignis [Punkt.01】
self.exec_btn_ctrl.Bind(wx.EVT_LEFT_DOWN, self.on_exec_click)
#Doppelklick-Ereignis mit linker Maus binden [Punkt.03】
self.exec_btn_ctrl.Bind(wx.EVT_LEFT_DCLICK, self.on_doubleclick)
self.btn_sizer.Add(self.exec_btn_ctrl, 0, wx.ALIGN_CENTER, 5)
#Suspend-Taste
self.kill_btn_ctrl = wx.Button(self, wx.ID_ANY, u"Lange Unterbrechung der Logikverarbeitung", wx.DefaultPosition, wx.Size(200, 50), 0)
#Mit der linken Maustaste an das Ereignis binden
self.kill_btn_ctrl.Bind(wx.EVT_LEFT_DOWN, self.on_kill_click)
#Doppelklick-Ereignis mit linker Maus binden
self.kill_btn_ctrl.Bind(wx.EVT_LEFT_DCLICK, self.on_doubleclick)
#Der Ausgangszustand ist inaktiv
self.kill_btn_ctrl.Disable()
self.btn_sizer.Add(self.kill_btn_ctrl, 0, wx.ALIGN_CENTER, 5)
self.sizer.Add(self.btn_sizer, 0, wx.ALIGN_CENTER | wx.SHAPED, 0)
#Konsole [Punkt.06】
self.console_ctrl = ConsoleCtrl(self)
self.sizer.Add(self.console_ctrl, 1, wx.ALL | wx.EXPAND, 5)
#Druckausgabeziel ist Konsole [Punkt.05】
sys.stdout = self.console_ctrl
#Fortschrittsanzeige
self.gauge_ctrl = wx.Gauge(self, wx.ID_ANY, 100, wx.DefaultPosition, wx.DefaultSize, wx.GA_HORIZONTAL)
self.gauge_ctrl.SetValue(0)
self.sizer.Add(self.gauge_ctrl, 0, wx.ALL | wx.EXPAND, 5)
#Ereignisbindung [Punkt.05】
self.Bind(EVT_LONG_THREAD, self.on_exec_result)
self.SetSizer(self.sizer)
self.Layout()
#Anzeige in der Mitte des Bildschirms
self.Centre(wx.BOTH)
#Doppelklicken Sie auf den Ungültigmachungsprozess
def on_doubleclick(self, event: wx.Event):
self.timer.Stop()
logger.warning("Es wurde doppelt geklickt.", decoration=MLogger.DECORATION_BOX)
event.Skip(False)
return False
#Ausführung 1 Verarbeitung beim Klicken
def on_exec_click(self, event: wx.Event):
#Beginnen Sie mit einer leichten Verzögerung mit einem Timer (vermeiden Sie das Schlagen mit einem Doppelklick) [Punkt.04】
self.timer = wx.Timer(self, TIMER_ID)
self.timer.StartOnce(200)
self.Bind(wx.EVT_TIMER, self.on_exec, id=TIMER_ID)
#Unterbrechung 1 Klicken Sie auf Verarbeitung
def on_kill_click(self, event: wx.Event):
self.timer = wx.Timer(self, TIMER_ID)
self.timer.StartOnce(200)
self.Bind(wx.EVT_TIMER, self.on_kill, id=TIMER_ID)
#Ausführung verarbeiten
def on_exec(self, event: wx.Event):
self.timer.Stop()
if not self.worker:
#Klar Konsole
self.console_ctrl.Clear()
#Schaltfläche "Ausführen" deaktivieren
self.exec_btn_ctrl.Disable()
#Schaltfläche "Anhalten" aktiviert
self.kill_btn_ctrl.Enable()
#In einem anderen Thread ausführen [Punkt.09】
self.worker = LongLogicWorker(self, LongThreadEvent, self.loop_cnt_ctrl.GetValue(), self.multi_process_ctrl.GetValue())
self.worker.start()
event.Skip(False)
#Unterbrechen Sie die Verarbeitung
def on_kill(self, event: wx.Event):
self.timer.Stop()
if self.worker:
#Wenn die Taste im gestoppten Zustand gedrückt wird, stoppt sie
self.worker.stop()
logger.warning("Unterbricht die lange Logikverarbeitung.", decoration=MLogger.DECORATION_BOX)
#Arbeiterende
self.worker = None
#Schaltfläche "Ausführen" aktiviert
self.exec_btn_ctrl.Enable()
#Deaktivieren Sie die Suspend-Schaltfläche
self.kill_btn_ctrl.Disable()
#Fortschritt verbergen
self.gauge_ctrl.SetValue(0)
event.Skip(False)
#Die Verarbeitung nach langer Logik ist beendet
def on_exec_result(self, event: wx.Event):
# 【Point.12] Machen Sie das logische Ende explizit bekannt
self.sound_finish()
#Schaltfläche "Ausführen" aktiviert
self.exec_btn_ctrl.Enable()
#Deaktivieren Sie die Suspend-Schaltfläche
self.kill_btn_ctrl.Disable()
if not event.result:
event.Skip(False)
return False
self.elapsed_time += event.elapsed_time
logger.info("\n Bearbeitungszeit: %s", self.show_worked_time())
#Arbeiterende
self.worker = None
#Fortschritt verbergen
self.gauge_ctrl.SetValue(0)
def sound_finish(self):
#Ton der Endton
if os.name == "nt":
# Windows
try:
import winsound
winsound.PlaySound("SystemAsterisk", winsound.SND_ALIAS)
except Exception:
pass
def show_worked_time(self):
#Konvertieren Sie verstrichene Sekunden in Stunden, Minuten und Sekunden
td_m, td_s = divmod(self.elapsed_time, 60)
if td_m == 0:
worked_time = "{0:02d}Sekunden".format(int(td_s))
else:
worked_time = "{0:02d}Protokoll{1:02d}Sekunden".format(int(td_m), int(td_s))
return worked_time
Lassen Sie uns zunächst das Ereignis mit der linken Maustaste an die Schaltfläche und die auszuführende Methode binden. Es gibt verschiedene Möglichkeiten zum Binden, aber ich persönlich mag die Möglichkeit, GUI-Teile und -Ereignisse als "Parts.Bind (Ereignistyp, Auslösemethode)" zu binden, da dies leicht zu verstehen ist.
Wenn Sie das Ereignis mit der linken Maustaste aufnehmen und es so ausführen, wie es ist, wird es gleichzeitig mit dem Doppelklickereignis ausgelöst, und als Ergebnis wird das Doppelklickereignis ausgeführt. (In Bezug auf die Verarbeitung ist das Doppelklickereignis Teil des Einzelklickereignisses, sodass die beiden Ereignisse gleichzeitig ausgelöst werden.) Wenn Sie einen Timer in "on_exec_click" und "on_kill_click" einstellen, der mit einem einzigen Klick ausgelöst und mit einer leichten Verzögerung ausgeführt wird, wird zuerst die an das Doppelklickereignis gebundene Ausführungsmethode ausgeführt.
Binden Sie das linke Doppelklickereignis auf die gleiche Weise wie Punkt ①. Sie können Doppelverarbeitung verhindern, indem Sie hier Doppelklicks ausführen.
Das Doppelklick-Ereignis stoppt das an Punkt② ausgeführte Timer-Ereignis. Sie können jetzt das Doppelklicken deaktivieren.
Es wird nur ein Ereignis mit einem Klick ausgelöst … Das entsprechende Ereignis wird mit einer leichten Verzögerung ausgeführt
Wenn das Doppelklick-Ereignis ausgelöst wird … Das Single-Click-Ereignis wird nicht ausgeführt, da der Timer gestoppt ist.
print
ist ein Wrapper für sys.stdout.write
. Wenn Sie also das Ausgabeziel auf ein Konsolensteuerelement festlegen, befindet sich das Ausgabeziel von print
innerhalb des Steuerelements.
Also, was ist das Konsolensteuerelement? Es ist eine Unterklasse von wx.TextCtrl
.
ConsoleCtrl.py
# -*- coding: utf-8 -*-
#
import wx
from utils.MLogger import MLogger # noqa
logger = MLogger(__name__)
class ConsoleCtrl(wx.TextCtrl):
def __init__(self, parent):
#Mehrere Zeilen zulässig, schreibgeschützt, kein Rand, mit vertikalem Bildlauf, mit horizontalem Bildlauf, mit Schlüsselereigniserfassung
super().__init__(parent, wx.ID_ANY, wx.EmptyString, wx.DefaultPosition, wx.Size(-1, -1), \
wx.TE_MULTILINE | wx.TE_READONLY | wx.BORDER_NONE | wx.HSCROLL | wx.VSCROLL | wx.WANTS_CHARS)
self.SetBackgroundColour(wx.SystemSettings.GetColour(wx.SYS_COLOUR_3DLIGHT))
#Tastaturereignisbindung
self.Bind(wx.EVT_CHAR, lambda event: self.on_select_all(event, self.console_ctrl))
#Ausgabeverarbeitung des Konsolenteils [Punkt.07】
def write(self, text):
try:
wx.CallAfter(self.AppendText, text)
except: # noqa
pass
#Alle Auswahlvorgänge des Konsolenteils [Punkt.08】
def on_select_all(event, target_ctrl):
keyInput = event.GetKeyCode()
if keyInput == 1: # 1 stands for 'ctrl+a'
target_ctrl.SelectAll()
event.Skip()
Rufen Sie die AppendText-Methode mit CallAfter auf, vorausgesetzt, sie wird von einem anderen Logik-Thread als dem GUI-Thread aufgerufen. Dadurch wird auch die Druckausgabe des Logik-Threads stabilisiert.
Wenn es Buchstaben gibt, ist es die menschliche Saga, die Sie dazu bringt, sie zu kopieren. Daher wird der All-Selection-Prozess im All-Selection-Ereignis (Kombination von Tastaturereignissen) ausgeführt.
Ich bin endlich zum Hauptthema gekommen.
Führen Sie die Logikverarbeitung in LongLogicWorker
durch.
Referenzquelle: https://doloopwhile.hatenablog.com/entry/20090627/1275175850
LongLogicWorker.py
# -*- coding: utf-8 -*-
#
import os
import wx
import time
from worker.BaseWorker import BaseWorker, task_takes_time
from service.MOptions import MOptions
from service.LongLogicService import LongLogicService
class LongLogicWorker(BaseWorker):
def __init__(self, frame: wx.Frame, result_event: wx.Event, loop_cnt: int, is_multi_process: bool):
#Verarbeitungszeit
self.loop_cnt = loop_cnt
#Gibt an, ob in mehreren Prozessen ausgeführt werden soll
self.is_multi_process = is_multi_process
super().__init__(frame, result_event)
@task_takes_time
def thread_event(self):
start = time.time()
#Para und Optionsfüllung
# max_Der maximale Wert von Workern ist Python3.Basierend auf dem Standardwert von 8
options = MOptions(self.frame.version_name, self.frame.logging_level, self.loop_cnt, max_workers=(1 if not self.is_multi_process else min(32, os.cpu_count() + 4)))
#Ausführung des Logikdienstes
LongLogicService(options).execute()
#verstrichene Zeit
self.elapsed_time = time.time() - start
def post_event(self):
#Rufen Sie das Ereignis auf und führen Sie es aus, nachdem die Logikverarbeitung abgeschlossen ist [Punkt.11】
wx.PostEvent(self.frame, self.result_event(result=self.result and not self.is_killed, elapsed_time=self.elapsed_time))
LongLogicWorker
erbt von BaseWorker
.
BaseWorker.py
# -*- coding: utf-8 -*-
#
import wx
import wx.xrc
from abc import ABCMeta, abstractmethod
from threading import Thread
from functools import wraps
import time
import threading
from utils.MLogger import MLogger # noqa
logger = MLogger(__name__)
# https://wiki.wxpython.org/LongRunningTasks
# https://teratail.com/questions/158458
# http://nobunaga.hatenablog.jp/entry/2016/06/03/204450
class BaseWorker(metaclass=ABCMeta):
"""Worker Thread Class."""
def __init__(self, frame, result_event):
"""Init Worker Thread Class."""
#Übergeordnete GUI
self.frame = frame
#verstrichene Zeit
self.elapsed_time = 0
#Ereignis aufrufen, nachdem der Thread beendet wurde
self.result_event = result_event
#Fortschrittsanzeige
self.gauge_ctrl = frame.gauge_ctrl
#Erfolgreiche Bearbeitung
self.result = True
#Mit oder ohne Stoppbefehl
self.is_killed = False
#Fadenstart
def start(self):
self.run()
#Fadenstopp
def stop(self):
#Interrupt FLG einschalten
self.is_killed = True
def run(self):
#Thread-Ausführung
self.thread_event()
#Nachbearbeitungsausführung
self.post_event()
def post_event(self):
wx.PostEvent(self.frame, self.result_event(result=self.result))
@abstractmethod
def thread_event(self):
pass
# https://doloopwhile.hatenablog.com/entry/20090627/1275175850
class SimpleThread(Thread):
"""Ein Thread, der nur ein aufrufbares Objekt ausführt (z. B. eine Funktion)."""
def __init__(self, base_worker, acallable):
#Verarbeitung in einem anderen Thread
self.base_worker = base_worker
#Methode zum Bewegen im Funktionsdekorateur
self.acallable = acallable
#Ergebnis der Funktion Dekorateur
self._result = None
#Suspended FLG=Im AUS-Zustand initialisieren
super(SimpleThread, self).__init__(name="simple_thread", kwargs={"is_killed": False})
def run(self):
self._result = self.acallable(self.base_worker)
def result(self):
return self._result
def task_takes_time(acallable):
"""
Funktionsdekorateur [Punkt.10】
Während der Ausführung der ursprünglichen Verarbeitung von acallable in einem anderen Thread
Fenster wx aktualisieren.Rufen Sie weiterhin YieldIfNeeded an
"""
@wraps(acallable)
def f(base_worker):
t = SimpleThread(base_worker, acallable)
#Dämon tötet Kinder, wenn Eltern sterben
t.daemon = True
t.start()
#Aktualisieren Sie die Fensterzeichnung für die gesamte Lebensdauer des Threads
while t.is_alive():
#Drehen Sie die Fortschrittsanzeige
base_worker.gauge_ctrl.Pulse()
#Aktualisieren Sie gegebenenfalls das Fenster
wx.YieldIfNeeded()
#Warte ein wenig
time.sleep(0.01)
if base_worker.is_killed:
# 【Point.23] Wenn der Anrufer einen Stoppbefehl ausgibt, Sie(GUI)Beendigungsbefehl für alle Threads außer
for th in threading.enumerate():
if th.ident != threading.current_thread().ident and "_kwargs" in dir(th):
th._kwargs["is_killed"] = True
break
return t.result()
return f
Dies ist eine Geschichte, die ich bereits von der Referenzseite erhalten habe, aber ich werde die Zeichnung des GUI-Threads weiterhin aktualisieren, während ich einen anderen Thread mit dem Funktionsdekorator ausführe.
Führen Sie acallable
in SimpleThread
aus.
Zu diesem Zeitpunkt besteht der Grund für das Halten von "BaseWorker" darin, die Suspendierungsflagge zu übergeben.
Während "SimpleThread" aktiv ist, aktualisiert der GUI-Thread nur die Zeichnung und akzeptiert Unterbrechungen.
(Dieser Bereich wird später angezeigt)
Wenn der Worker im Voraus initialisiert wurde, wurde das Aufrufereignis nach der Verarbeitung übergeben. Führen Sie es daher mit "wx.PostEvent" aus. Dadurch kehren Sie zur Verarbeitung in der GUI zurück.
Nicht alle Benutzer sitzen am PC fest, bis die Logik fertig ist. Daher versuche ich bei der Dimensionierung, ein INFO-Geräusch zu erzeugen, damit es leicht zu verstehen ist, wenn es fertig ist. Abhängig von der Windows-Umgebung und anderen Umgebungen wie Linux kann ein Fehler auftreten. Es scheint daher besser, ihn nur dann klingen zu lassen, wenn er mit try-without erklingen kann. Übrigens, wenn Sie die verstrichene Zeit angeben, wird das Gefühl, wie lange es gedauert hat, quantifiziert, sodass ich der Meinung bin, dass es leicht zu verstehen ist.
LongLogicService.py
# -*- coding: utf-8 -*-
#
import logging
from time import sleep
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor
from service.MOptions import MOptions
from utils.MException import MLogicException, MKilledException
from utils.MLogger import MLogger # noqa
logger = MLogger(__name__)
class LongLogicService():
def __init__(self, options: MOptions):
self.options = options
def execute(self):
logging.basicConfig(level=self.options.logging_level, format="%(message)s [%(module_name)s]")
# 【Point.13] versuche das ganze-Fügen Sie Ausnahme ein und geben Sie den Fehlerinhalt aus
try:
#Es ist in Ordnung, die Logik normal zu setzen
self.execute_inner(-1)
logger.info("--------------")
#Es ist in Ordnung, mit parallelen Aufgaben zu verteilen
futures = []
# 【Point.14] Geben Sie der parallelen Aufgabe einen Namen
with ThreadPoolExecutor(thread_name_prefix="long_logic", max_workers=self.options.max_workers) as executor:
for n in range(self.options.loop_cnt):
futures.append(executor.submit(self.execute_inner, n))
#【Point.15] Parallele Tasks warten auf den Abschluss, nachdem sie in einem Stapel ausgegeben wurden
concurrent.futures.wait(futures, timeout=None, return_when=concurrent.futures.FIRST_EXCEPTION)
for f in futures:
if not f.result():
return False
logger.info("Ende der langen Logikverarbeitung", decoration=MLogger.DECORATION_BOX, title="Logike Ende")
return True
except MKilledException:
#Bei Beendigung durch Suspend-Option wird nur das Ergebnis so zurückgegeben, wie es ist
return False
except MLogicException as se:
#Unvollständiger Datenfehler
logger.error("Es endete mit Daten, die nicht verarbeitet werden können.\n\n%s", se.message, decoration=MLogger.DECORATION_BOX)
return False
except Exception as e:
#Andere Fehler
logger.critical("Der Prozess endete mit einem unbeabsichtigten Fehler.", e, decoration=MLogger.DECORATION_BOX)
return False
finally:
logging.shutdown()
def execute_inner(self, n: int):
for m in range(5):
logger.info("n: %s - m: %s", n, m)
sleep(1)
return True
Da die Threads getrennt sind und Sie den Fehler nicht richtig ausschließen, können Sie plötzlich enden, obwohl Sie nicht wissen, was es ist. Es ist schwer, danach zu jagen, also lassen Sie es uns ausschließen und protokollieren.
Bei der Ausführung paralleler Aufgaben ist das Debuggen einfacher, wenn Sie ein Präfix hinzufügen, damit Sie leicht verstehen, welcher Prozess das Problem ist.
Parallele Aufgaben werden zuerst mit "executor.submit" ausgegeben und dann mit "concurrent.futures.wait" gewartet, bis die gesamte Verarbeitung abgeschlossen ist. Zu diesem Zeitpunkt wird die Option "concurrent.futures.FIRST_EXCEPTION" hinzugefügt, damit die Verarbeitung unterbrochen wird, wenn eine Ausnahme auftritt.
MLogger.py
# -*- coding: utf-8 -*-
#
from datetime import datetime
import logging
import traceback
import threading
from utils.MException import MKilledException
# 【Point.16] Implementieren Sie Ihren eigenen Logger
class MLogger():
DECORATION_IN_BOX = "in_box"
DECORATION_BOX = "box"
DECORATION_LINE = "line"
DEFAULT_FORMAT = "%(message)s [%(funcName)s][P-%(process)s](%(asctime)s)"
DEBUG_FULL = 2
TEST = 5
TIMER = 12
FULL = 15
INFO_DEBUG = 22
DEBUG = logging.DEBUG
INFO = logging.INFO
WARNING = logging.WARNING
ERROR = logging.ERROR
CRITICAL = logging.CRITICAL
total_level = logging.INFO
is_file = False
outout_datetime = ""
logger = None
#Initialisieren
# 【Point.17] Sie können den minimalen Ausgangspegel für jedes Modul definieren
def __init__(self, module_name, level=logging.INFO):
self.module_name = module_name
self.default_level = level
#Logger
self.logger = logging.getLogger("PyLogicSample").getChild(self.module_name)
#Standardausgabehandler
sh = logging.StreamHandler()
sh.setLevel(level)
self.logger.addHandler(sh)
# 【Point.18] Bereiten Sie eine Protokollmethode mit einer niedrigeren Ebene als Debug vor
def test(self, msg, *args, **kwargs):
if not kwargs:
kwargs = {}
kwargs["level"] = self.TEST
kwargs["time"] = True
self.print_logger(msg, *args, **kwargs)
def debug(self, msg, *args, **kwargs):
if not kwargs:
kwargs = {}
kwargs["level"] = logging.DEBUG
kwargs["time"] = True
self.print_logger(msg, *args, **kwargs)
def info(self, msg, *args, **kwargs):
if not kwargs:
kwargs = {}
kwargs["level"] = logging.INFO
self.print_logger(msg, *args, **kwargs)
def warning(self, msg, *args, **kwargs):
if not kwargs:
kwargs = {}
kwargs["level"] = logging.WARNING
self.print_logger(msg, *args, **kwargs)
def error(self, msg, *args, **kwargs):
if not kwargs:
kwargs = {}
kwargs["level"] = logging.ERROR
self.print_logger(msg, *args, **kwargs)
def critical(self, msg, *args, **kwargs):
if not kwargs:
kwargs = {}
kwargs["level"] = logging.CRITICAL
self.print_logger(msg, *args, **kwargs)
#Tatsächliche Ausgabe
def print_logger(self, msg, *args, **kwargs):
#【Point.22] FLG auf dem aktuell laufenden Thread anhalten=Wenn ON gesetzt ist, tritt ein Unterbrechungsfehler auf.
if "is_killed" in threading.current_thread()._kwargs and threading.current_thread()._kwargs["is_killed"]:
#Wenn ein Stoppbefehl ausgegeben wird, liegt ein Fehler vor
raise MKilledException()
target_level = kwargs.pop("level", logging.INFO)
#Ausgabe nur, wenn sowohl die App- als auch die Modulprotokollstufe erfüllt sind
if self.total_level <= target_level and self.default_level <= target_level:
if self.is_file:
for f in self.logger.handlers:
if isinstance(f, logging.FileHandler):
#Löschen Sie alle vorhandenen Dateihandler
self.logger.removeHandler(f)
#Wenn eine Datei ausgegeben wird, Handlerzuordnung
#Dateiausgabehandler
fh = logging.FileHandler("log/PyLogic_{0}.log".format(self.outout_datetime))
fh.setLevel(self.default_level)
fh.setFormatter(logging.Formatter(self.DEFAULT_FORMAT))
self.logger.addHandler(fh)
#Zum Namen des Ausgabemoduls hinzugefügt
extra_args = {}
extra_args["module_name"] = self.module_name
#Protokolldatensatzgenerierung
if args and isinstance(args[0], Exception):
# 【Point.19] Wenn eine Ausnahme empfangen wird, wird eine Stapelverfolgung ausgegeben.
log_record = self.logger.makeRecord('name', target_level, "(unknown file)", 0, "{0}\n\n{1}".format(msg, traceback.format_exc()), None, None, self.module_name)
else:
log_record = self.logger.makeRecord('name', target_level, "(unknown file)", 0, msg, args, None, self.module_name)
target_decoration = kwargs.pop("decoration", None)
title = kwargs.pop("title", None)
print_msg = "{message}".format(message=log_record.getMessage())
# 【Point.20] Dekorieren Sie Protokollnachrichten mit Parametern
if target_decoration:
if target_decoration == MLogger.DECORATION_BOX:
output_msg = self.create_box_message(print_msg, target_level, title)
elif target_decoration == MLogger.DECORATION_LINE:
output_msg = self.create_line_message(print_msg, target_level, title)
elif target_decoration == MLogger.DECORATION_IN_BOX:
output_msg = self.create_in_box_message(print_msg, target_level, title)
else:
output_msg = self.create_simple_message(print_msg, target_level, title)
else:
output_msg = self.create_simple_message(print_msg, target_level, title)
#Ausgabe
try:
if self.is_file:
#Wenn eine Datei ausgegeben wird, generieren Sie den Datensatz neu und geben Sie sowohl die Konsole als auch die GUI aus
log_record = self.logger.makeRecord('name', target_level, "(unknown file)", 0, output_msg, None, None, self.module_name)
self.logger.handle(log_record)
else:
# 【Point.21] Der Logik-Thread wird für Druck und Logger separat ausgegeben
print(output_msg)
self.logger.handle(log_record)
except Exception as e:
raise e
def create_box_message(self, msg, level, title=None):
msg_block = []
msg_block.append("■■■■■■■■■■■■■■■■■")
if level == logging.CRITICAL:
msg_block.append("■ **CRITICAL** ")
if level == logging.ERROR:
msg_block.append("■ **ERROR** ")
if level == logging.WARNING:
msg_block.append("■ **WARNING** ")
if level <= logging.INFO and title:
msg_block.append("■ **{0}** ".format(title))
for msg_line in msg.split("\n"):
msg_block.append("■ {0}".format(msg_line))
msg_block.append("■■■■■■■■■■■■■■■■■")
return "\n".join(msg_block)
def create_line_message(self, msg, level, title=None):
msg_block = []
for msg_line in msg.split("\n"):
msg_block.append("■■ {0} --------------------".format(msg_line))
return "\n".join(msg_block)
def create_in_box_message(self, msg, level, title=None):
msg_block = []
for msg_line in msg.split("\n"):
msg_block.append("■ {0}".format(msg_line))
return "\n".join(msg_block)
def create_simple_message(self, msg, level, title=None):
msg_block = []
for msg_line in msg.split("\n"):
# msg_block.append("[{0}] {1}".format(logging.getLevelName(level)[0], msg_line))
msg_block.append(msg_line)
return "\n".join(msg_block)
@classmethod
def initialize(cls, level=logging.INFO, is_file=False):
# logging.basicConfig(level=level)
logging.basicConfig(level=level, format=cls.DEFAULT_FORMAT)
cls.total_level = level
cls.is_file = is_file
cls.outout_datetime = "{0:%Y%m%d_%H%M%S}".format(datetime.now())
Der vielleicht genialste Teil ist der Logger-Teil. Es ist für Benutzer einfacher, das Protokoll in einem bestimmten Format auszugeben, aber es ist sehr mühsam, es einzeln zu definieren. Sie können Nachrichten wie Kästchen und Rahmen mit einer einzigen Flagge dekorieren. Und vor allem wird der Logger verwendet, um das Unterbrechungsflag zu beurteilen. (Details werden später beschrieben)
Wenn Sie für jedes Modul den Mindestausgangspegel definieren, können Sie insbesondere das Debug-Protokoll von Dienstprogrammmethoden unterdrücken. Das physische Löschen oder Auskommentieren des Debug-Protokolls kann schwierig sein, um zu überprüfen, ob ein Problem vorliegt. Durch Erhöhen oder Verringern des Mindestpegels für jedes Modul können Sie den Ausgabeprotokollpegel steuern, was zu einem einfacheren Debugging führt.
Obwohl es mit 17 gepaart ist, ist es einfacher, die Ausgabe durch Vorbereiten einer Low-Level-Methode zu unterdrücken.
Dies ist hauptsächlich nützlich, wenn Sie eine nicht behandelte Ausnahme erhalten. Da der Logger sowohl vom GUI-Thread als auch vom Logik-Thread verarbeitet wird, muss die Ausgabe an der Quelle des Logik-Threads nicht angepasst werden.
Da wir das Konsolensteuerelement in ein reguläres Textsteuerelement umgewandelt haben, haben wir aus Gründen der Übersichtlichkeit viele Nachrichtenblockierungen verwendet. Da die Anzahl der Nachrichten variabel ist, konnte keine feste Zeichenfolge zugewiesen werden, sodass für jeden Parameter ein Blockieren aufgerufen wird. Ich denke, es gibt eine Möglichkeit, die Methode beim Aufrufer anzugeben, aber ich denke, dass dies in Bezug auf die Bedeutung einfacher zu verwalten ist. Selbst beim Dekorieren von Text ist die Menge an Code geringer, wenn er auf diese Weise an einer Stelle behandelt wird, anstatt vom Aufrufer getrennt zu werden.
Zum Drucken auf dem Konsolensteuerelement benötigen Sie die Ausgabe von "print", und zum Drucken in den Stream benötigen Sie die Ausgabe von "logger.handle". Beide Nachrichten geben die gleichen Informationen aus, und Informationen wie das Ausgabemodul und die Ausgabezeit werden der Ausgabe zum Stream hinzugefügt, um das Verfolgen zu erleichtern.
Hier war ich am meisten besorgt ... Im Folgenden sind einige der Thread-ähnlichen Praktiken von Python aufgeführt.
Referenz: https://stackoverflow.com/questions/323972/is-there-any-way-to-kill-a-thread
Selbst wenn ich die Unterbrechungsparameter intern betrachte, möchte ich sie nicht für jeden Logikprozess sehen und ich möchte die Parameter nicht herumtragen ... Ist es nicht der Logger, der immer eine Logik durchläuft? Wenn Sie es also im Logger sehen und im GUI-Thread ändern können, handelt es sich um den "aktuellen Thread" ... Indem ich das sagte, wurde es so.
Ich weiß nicht, ob es eine gute Methode ist, aber ich mag es, weil der Logikprozess keine Unterbrechungen berücksichtigt.
Das Suspend-FLG wird im Funktionsdekorator von BaseWorker
eingestellt.
Anhalten aller lebenden Threads außer GUI-Thread Wenn Sie FLG aktivieren, können Sie die Unterbrechung sehen, indem Sie sich einen beliebigen Thread ansehen.
Wenn Sie nun versuchen, das Protokoll auszugeben, tritt ein Fehler auf und Sie kehren zur GUI zurück.
Normale Beendigungsroute
Angehaltene Endroute
Nachdem wir die Umgebung erstellt haben, lassen wir sie von VS Code aus ausführen.
Geben Sie im Feld "Python-Pfad" des Arbeitsbereichs den vollständigen Pfad von "Anaconda> envs> Entwicklungsumgebung> python.exe" an.
Verwenden Sie launch
, um die Ausführung der exe anzugeben.
{
"folders": [
{
"path": "src"
}
],
"settings": {
"python.pythonPath": "C:\\Development\\Anaconda3\\envs\\pytest_env\\python.exe"
},
"launch": {
"version": "0.2.0",
"configurations": [
{
"name": "Python: debug",
"type": "python",
"request": "launch",
"program": "${workspaceFolder}/executor.py",
"console": "integratedTerminal",
"pythonPath": "${command:python.interpreterPath}",
"stopOnEntry": false,
"args": [
// "--verbose", "1", //Minimum
// "--verbose", "2", // DEBUG_FULL
// "--verbose", "15", // FULL
"--verbose", "10", // TEST
// "--verbose", "20", // INFO
]
}
]
}
}
Sie können die GUI jetzt über VS Code starten.
Ich habe viel Code zusammengestellt, aber am Ende muss ich es zu einer Exe machen. Hier sind die Stapel und Konfigurationsdateien, mit denen PythonExe erstellt wird.
pyinstaller64.bat
@echo off
rem ---
rem ---Exe generieren
rem ---
rem ---Ändern Sie das aktuelle Verzeichnis in das Ausführungsziel
cd /d %~dp0
cls
rem ---Führen Sie nach dem Wechsel zur Release-Umgebung pyinstaller aus
rem ---Kehren Sie zur Entwicklungsumgebung zurück, wenn Sie fertig sind
activate pytest_release && pyinstaller --clean pytest64.spec && activate pytest_env
pytest64.spec
# -*- coding: utf-8 -*-
# -*- mode: python -*-
#PythonExe-Beispielversion mit 64 Bit
block_cipher = None
a = Analysis(['src\\executor.py'],
pathex=[],
binaries=[],
datas=[],
#Import versteckter Bibliotheken
hiddenimports=['wx._adv', 'wx._html', 'pkg_resources.py2_warn'],
hookspath=[],
runtime_hooks=[],
#Auszuschließende Bibliotheken
excludes=['mkl','libopenblas'],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False)
pyz = PYZ(a.pure, a.zipped_data,
cipher=block_cipher)
exe = EXE(pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
#App Name
name='PythonExeSample.exe',
#Gibt an, ob das Debug-Protokoll beim Erstellen einer Exe angezeigt werden soll
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
#Gibt an, ob die Konsole angezeigt werden soll
console=False )
Wenn Sie sich an Exe halten möchten, führen Sie einfach den Stapel aus, wechseln Sie zur Release-Umgebung, führen Sie "pyinstaller" aus und kehren Sie dann zur Entwicklungsumgebung zurück. Jetzt müssen Sie sich keine Sorgen mehr machen, dass Sie Ihrer Release-Umgebung versehentlich zusätzliche Bibliotheken hinzufügen.
Die Spezifikationsdatei ist die Einstellungsdatei von pyinstaller
, aber es ist die Einstellung, die durch die Kommentarzeile hinzugefügt wird.
hiddenimports
pyinstaller
bündelt grundsätzlich automatisch die im Code aufgerufenen Bibliotheken, aber es gibt einige Bibliotheken, die nicht so aufgenommen werden können, wie sie sind.
Der explizite Import ist "versteckter Import".
Die einzige Möglichkeit, dies zu finden, besteht darin, "debug = False" unten in "True" zu ändern und nach dem Teil zu suchen, der den Fehler verursacht ... Ich denke, es ist eine sehr einfache Aufgabe.
excludes
Im Gegenteil, wenn Sie es ausschließen möchten, weil die Dateigröße groß wird, wenn es zusammen gebündelt wird, geben Sie es mit "ausschließen" an. In diesem Fall werden "mkl" und "libopenblas" unter Bezugnahme auf https://www.reddit.com/r/pygame/comments/aelypb/why_is_my_pyinstaller_executable_180_mb_large/ ausgeschlossen. Die fertige Exe ist ca. 30M. (Auch wenn ausgeschlossen, diese Größe ...
Ich kann es hinzufügen, wenn sich meine Energie erholt. Wie geht es Ihnen hier mit der VMD-Dimensionierung? Fragen sind ebenfalls willkommen.
Der gesamte oben genannte Code finden Sie unter https://github.com/miu200521358/PythonExeSample. Wenn Sie interessiert sind, gabeln Sie sich bitte und schauen Sie sich den Inhalt an.