Запрос глубины очереди событий root.main()/Tkinter/TCL

У меня есть приложение, написанное с использованием пользовательского интерфейса tkinter, которое выполняет реальные действия через стек GPIO.

В качестве примера представьте, что мое приложение имеет датчик температуры, который проверяет температуру в помещении, а затем имеет внешний аппаратный интерфейс с системой кондиционирования воздуха для регулировки выходной температуры системы HVAC. Это делается простым способом через стек GPIO.

Приложение имеет некоторые действия типа «критически важные для безопасности», которые оно может выполнять. Существует множество аппаратных блокировок, но я по-прежнему не хочу, чтобы приложение вышло из строя и заставило кондиционер работать на полную мощность и превратить мой теоретический дом в морозильную камеру.

Проблема всегда в том... как преобразовать приложение, управляемое событиями графического пользовательского интерфейса, в приложение с графическим интерфейсом, в котором также выполняется еще один цикл без:

1. Нарушение дизайна tkinter/tcl, помещая все в «пока (true): делай что-то» и добавляя к нему root.update().
2. Добавление потоков в приложение, что сложно и потенциально небезопасно.

Итак, я остановился на следующей архитектуре:

импортировать tkinter как tk

def main(args):
    
    # Build the UI for the user
    root = tk.Tk()
    buttonShutDown = tk.Button(root, text = "PUSH ME!")#, command=sys.exit(0))
    buttonShutDown.pack()
    
    # Go into the method that looks at inputs and outputs
    monitorInputsAndActionOutputs(root)
    
    # The main tkinter loop. NOTE, there are about 9000 articles on why you should NOT just shove everyhtnig in a loop and call root.update(). This is the corect way of doing it! Use root.after() calls!
    root.mainloop()

def monitorInputsAndActionOutputs(root): 

    print ("checked the room temperature")
    print ("adjusted the airconditioning to make comfy")
    root.after(100, monitorInputsAndActionOutputs, root) # readd one event of the monitorInputsAndActionOutputs process to the tkinter / tcl queue

    return None

if __name__ == '__main__':
    import sys
    sys.exit(main(sys.argv))

На самом деле это прекрасно работает, и теоретически будет работать вечно, и каждый будет чувствовать себя комфортно(TM).

Тем не менее, я вижу потенциальную проблему в том, что приложение имеет приличное количество методов, выполняющих различные настройки стиля PID с замкнутым циклом (чтение, размышление, изменение, итерация), и я обеспокоен тем, что очередь событий tkinter начнет засоряться обработкой после (), и вся система либо замедлится, либо полностью сломается, либо заморозит пользовательский интерфейс, либо произойдет комбинация всех трех факторов.

Итак, я хочу добавить что-то вроде (грубый пример кода, это, конечно, не работает):

def (checkQueueDepth):
   
   myQueue = root.getQueueLength()

   if (myQueue > 100 items):
       doSomethingToAlertSomeone()
       deprioritiseNonCriticalEventsTillThingsCalmTFDown()

Чтобы реализовать что-то подобное, мне нужно программно получить глубину очереди root/tkinter.

После продолжительного почесывания бороды пробираюсь через

1. документы Ткинтера
2. Документы TCL/TK
3. Исходный код tkinter
4. Остальная часть stackoverflow
5. Интернет
6. Холодильник (для перекуса)

Я установил, что внешний модуль не может запросить длину очереди tk, который основан на библиотеке tcl и, вероятно, реализует или расширяет их уведомитель .

Судя по коду tkinter, также не похоже, что существует какой-либо частный метод или объект/переменная, к которому я могу проникнуть через парадную дверь и допросить.

В-третьих, единственное «решение», которое это выглядит, — это потенциально реализовать новую систему уведомлений для TCL, а затем расширить ее функциональность для обеспечения этой функции. У меня нет шансов сделать это, у меня нет ни знаний, ни времени, и я бы скорее переписал все свое приложение на другом языке.

Поскольку это невозможно сделать «правильным» способом, я подумал о способе «мы не собираемся называть это неправильным, но вполне могли бы и быть».

То, что я придумал, работает примерно так:

def checkCodeLoops(root):
    
    # The maximum time we will accept the queue processing to blow out to is 3 seconds.
    maxDelayInQueueResponse = datetime.timedelta(seconds = 3)
    
    # Take a note of the time now as well as the last time this method was run
    lastCheckTime = checkTime
    checkTime = datetime.datetime.now()
    
    # If the time between this invocation of the checkCodeLoops method and the last one is greater than the max allowable time, panic!
    if (checkTime - lastCheckTime > maxDelayInQueueResponse):
        doSomethingToReduceLoad()
        lockUISoUserCantAddMoreWorkToTheQueue()
    
    # reque this method so that we can check again in 100ms (theoretically in 100ms at least - if the queue's flooded then this might never fire)
    root.after_idle(100, monitorInputsAndActionOutputs, root)
    
    return None

В общем, это дерьмовый способ сделать это. По сути, вы добавляете работу в очередь, чтобы увидеть, занята ли очередь (глупо), а также есть еще одна проблема: если очередь переполнена, то этот код на самом деле не запустится, потому что любые события, добавленные в процесс after_idle(), являются обрабатывается только тогда, когда в очереди нет after() и других элементов с более высоким приоритетом.

Поэтому я мог бы изменить его на root.after вместо root.after_idle, и теоретически это помогло бы, но у нас все еще остается проблема с использованием события в очереди, чтобы проверить, не работает ли очередь, что довольно глупо.

Если бы у меня был способ проверить глубину очереди, я мог бы начать реализовывать стратегии управления нагрузкой ДО того, как дело дошло до стадии паники.

Очень хотелось бы услышать, есть ли у кого-нибудь лучший способ сделать это.