时间戳

首页 > 解决方案 > 为什么我的代码在我还没有提出时在自定义异常类中执行

python - 为什么我的代码在我还没有提出时在自定义异常类中执行

问题描述

我的项目中有 4 个文件：Tools.py、ProjectGUI.py、ProjectExceptions.py 和 ProjectMain.py

大多数实用程序和代码都写在 Tools 文件中，而 ProjectMain 只是从 Tools 中调用函数。我正在使用一个简单的带有 tkinter 的 GUI，写在 ProjectGUI.py 文件中。我指定了一个按钮来执行 ProjectMain 的主要代码。当我自己运行 ProjectMain 并尝试从 ProjectExceptions.py 中引发异常时，一切都按预期工作。但是，当我尝试使用 GUI 执行代码并引发异常时，它不起作用。在 Exception 类中编写的代码不会执行，我的程序只是冻结。

这是 ProjectExceptions.py 文件中的示例：

from tkinter.messagebox import showerror


class OffsetNotStable(BaseException):
    def __init__(self):
        showerror(title='Error', message='Not stable')


class VoltageNotWithinTolerance(BaseException):
    def __init__(self):
        showerror(title='Error', message='Voltage not correct')

我在 ProjectMain 中有 try/except。谁能告诉我我做错了什么？

编辑：这是来自 main 的代码片段：

class OffsetStableThread(threading.Thread):
    def __init__(self, dwf_ain, tolerance, name, ch, th_q):
        threading.Thread.__init__(self)
        self.name = name
        self.ch = ch
        self.dwf_ain = dwf_ain
        self.tolerance = tolerance
        self.th_q = th_q
        self.exc = None

    def run(self):
        try:
            # print("Starting " + self.name)
            Tools.offset_stable(self.dwf_ain,
                                self.ch,
                                tolerance=self.tolerance,
                                debug_mode=False,
                                th_q=self.th_q)
            # print("Exiting " + self.name)
        except BaseException as e:
            self.exc = e

    def my_join(self):
        threading.Thread.join(self)
        if self.exc:
            raise self.exc


def main():
    thread_to_exit = Queue()
    thread_to_exit.put(False)
    wd_const = DeviceData.Tolerances()
    hdwf = dwf.Dwf(idxCfg=1)
    dwf_ain = dwf.DwfAnalogIn(hdwf)
    try:
        offset_stable_thread_ch1 = OffsetStableThread(dwf_ain,
                                                      wd_const.offset_stable,
                                                      name='Offset stable channel 1',
                                                      ch=0,
                                                      th_q=thread_to_exit)
        offset_stable_thread_ch2 = OffsetStableThread(dwf_ain,
                                                      wd_const.offset_stable,
                                                      name='Offset stable channel 2',
                                                      ch=1,
                                                      th_q=thread_to_exit)
        offset_stable_thread_ch1.setDaemon(True)
        offset_stable_thread_ch2.setDaemon(True)
        offset_stable_thread_ch1.start()
        offset_stable_thread_ch2.start()
        offset_stable_thread_ch1.my_join()
        offset_stable_thread_ch2.my_join()

    except ProjectExceptions.OffsetNotStable:
        hdwf.close()
        return
    except ProjectExceptions.VoltageNotWithinTolerance:
        hdwf.close()
        return

这是从工具调用的函数：

def offset_stable(dwf_ain, channel, tolerance, debug_mode=False, th_q=None):
    dwf_ain.frequencySet(20e6)
    dwf_ain.bufferSizeSet(4000)
    dwf_ain.channelEnableSet(channel, True)
    dwf_ain.channelRangeSet(channel, 50)

    dc_old = 0
    count = -1
    for tries in range(tolerance['Max tries quota']):

        if th_q:
            if th_q.get():
                return
            else:
                th_q.put(False)

        count += 1
        dwf_ain.configure(False, True)
        while True:
            if dwf_ain.status(True) == dwf_ain.STATE.DONE:
                break
            time.sleep(0.1)

        rgdSamples = dwf_ain.statusData(channel, 4000)
        dc = sum(rgdSamples) / len(rgdSamples)
        if debug_mode:
            print(str(tries) + f"Channel {channel}: DC: " + str(dc) + "V")

        if count < tolerance['Skip entries']:
            continue
        count = 0
        slope = dc - dc_old
        # if debug_mode:
        #     print(f'slope = {slope}\n')
        dc_old = dc
        if slope > tolerance['Slope tolerance']:
            continue
        if debug_mode:
            print('\n', tries)
        return dwf_ain
    th_q.put(True)
    raise ProjectExceptions.OffsetNotStable

当从不同的线程引发异常时，我正在使用传递了 th_q 变量的 Queue 以退出线程。

标签： pythonpython-3.xtkinterexception

解决方案

我没有在函数中引发异常，而是返回了一个带有异常名称的字符串，并在线程类的 my_join 方法中引发了异常。

def offset_stable(dwf_ain, channel, tolerance, debug_mode=False, th_q=None):
    dwf_ain.frequencySet(20e6)
    dwf_ain.bufferSizeSet(4000)
    dwf_ain.channelEnableSet(channel, True)
    dwf_ain.channelRangeSet(channel, 50)

    dc_old = 0
    count = -1
    for tries in range(tolerance['Max tries quota']):

        if th_q:
            if th_q.get():
                return
            else:
                th_q.put(False)

        count += 1
        dwf_ain.configure(False, True)
        while True:
            if dwf_ain.status(True) == dwf_ain.STATE.DONE:
                break
            time.sleep(0.1)

        rgdSamples = dwf_ain.statusData(channel, 4000)
        dc = sum(rgdSamples) / len(rgdSamples)
        if debug_mode:
            print(str(tries) + f"Channel {channel}: DC: " + str(dc) + "V")

        if count < tolerance['Skip entries']:
            continue
        count = 0
        slope = dc - dc_old
        # if debug_mode:
        #     print(f'slope = {slope}\n')
        dc_old = dc
        if slope > tolerance['Slope tolerance']:
            continue
        if debug_mode:
            print('\n', tries)
        return dwf_ain
    th_q.put(True)
    return 'OffsetNotStable'

class OffsetStableThread(threading.Thread):
    def __init__(self, dwf_ain, tolerance, name, ch, th_q):
        threading.Thread.__init__(self)
        self.name = name
        self.ch = ch
        self.dwf_ain = dwf_ain
        self.tolerance = tolerance
        self.th_q = th_q
        self.offset = None

    def run(self):
        # print("Starting " + self.name)
        self.offset = Tools.offset_stable(self.dwf_ain,
                                          self.ch,
                                          tolerance=self.tolerance,
                                          debug_mode=False,
                                          th_q=self.th_q)
        # print("Exiting " + self.name)

    def my_join(self):
        threading.Thread.join(self)
        if self.offset == 'OffsetNotStable':
            raise TesterExceptions.OffsetNotStable

推荐阅读