J'écris un programme de contrôle qui a besoin de plusieurs minuteries configurables et redémarrables qui s'exécutent dans leur propre thread. À l'origine, j'ai implémenté cela avec des fonctions personnalisées et des noms d'événements personnalisés pour tous les minuteurs. Parce que c'est une approche vraiment moche, je fais une réécriture avec une approche plus proche de l'OO. Veuillez considérer l'exemple minimum ci-dessous qui fait tout ce dont j'ai besoin avec une exception pour laquelle je n'ai pas pu trouver de solution : je ne peux pas accéder aux attributs d'instance de classe requis (j'espère que j'utilise la bonne terminologie ici).

L'instruction défaillante est la ligne 74 et commentée. Je soupçonne un problème avec la façon dont je passe les arguments et les kwargs, mais je me suis heurté à un mur de briques. Pouvez-vous voir où je me trompe et suggérer une solution?

#!/usr/bin/python3
# coding: utf-8
"""MRE; needs installation of sine.threads ReStartableThread: pip3 install sine.threads"""
import sys
import time
import threading
from sine.threads import ReStartableThread as Thread


class MyThread(Thread):
    """
    drop-in replacement of ReStartableThread with status method added
    """

    def __init__(self, name=None, event_name=None, target=None, group=None, args=(), kwargs=None):
        Thread.__init__(self)
        self.name = name
        self.event_name = event_name
        self.target = target
        self.args = args
        self.kwargs = kwargs
        super().__init__(name=name, event_name=event_name, target=target, group=group, args=args, kwargs=kwargs)

    def status(self, i):  # placeholder
        print('dummy status #', i)


def threader(f):
    """ a threading decorator; put @threaded above the callable that needs threading """
    def threading_function(*args, **kwargs):
        threading.Thread(name=f.__name__, target=f, args=args, kwargs=kwargs).start()

    return threading_function


@threader
def t0():
    time.sleep(3)
    t1.stop()
    t1.join()


class MyTimer:

    def __init__(self, name, kwargs):
        for key, value in kwargs.items():  # unpack kwargs and convert to variables
            setattr(self, key, value)

    def __call__(self, stop_event):
        i = 5
        while not stop_event.is_set() and i > 0:
            print('I: ', i)
            time.sleep(1)
            i -= 1
        if stop_event.is_set():
            print('timer is CANCELLED')
        else:
            print('timer is FINISHED')


def main():
    global t1
    name = 'timer'
    t1_kwargs = {'interval': 99, 'message': 'STARTING COUNTDOWN...'}
    t1 = MyThread(name=name, target=MyTimer(name, t1_kwargs))
    t1.status(1)
    # first countdown
    t1.start()
    """
    The next statement fails with "AttributeError: 'StoppableThread' object has no attribute 'interval'"
    where StoppableThread is a member of the sine.threads module.
    Requirement: access to attributes (like interval) of instances of MyTimer from main
    """
    # print(t1.interval)
    t0()
    t1.join()
    t1.status(2)
    # second countdown
    t1.start()
    t1.status(3)
    t1.join()
    t1.status(4)
    print('END')


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('\b\b\r')
        print("PROGRAM TERMINATED AT USER'S REQUEST")
        force_exit()
0
Alfred Elkerbout 15 oct. 2020 à 14:03

1 réponse

Meilleure réponse

Tout d'abord, vous essayez d'initialiser votre classe de base Thread deux fois, d'abord avec

Thread.__init__(self)

Puis avec:

super().__init__(name=name, event_name=event_name, target=target, group=group, args=args, kwargs=kwargs)

Vous devriez vous débarrasser de la première initialisation.

Votre problème est que vous attribuez t1 à votre instance MyThread :

t1 = MyThread(name=name, target=MyTimer(name, t1_kwargs))

Mais interval est un attribut de votre instance MyTimer, pas votre instance MyThread. Donc t1.interval échouera. Vous voulez:

t1 = MyTimer(name, t1_kwargs)
my_thread = MyThread(name, target=t1)
my_thread.status(1)
my_thread.start()
etc.

Vous avez juste besoin de garder droit vos deux instances différentes :

#!/usr/bin/python3
# coding: utf-8
"""MRE; needs installation of sine.threads ReStartableThread: pip3 install sine.threads"""
import sys
import time
import threading
from sine.threads import ReStartableThread as Thread


class MyThread(Thread):
    """
    drop-in replacement of ReStartableThread with status method added
    """

    def __init__(self, name=None, event_name=None, target=None, group=None, args=(), kwargs=None):
        self.name = name
        self.event_name = event_name
        self.target = target
        self.args = args
        self.kwargs = kwargs
        super().__init__(name=name, event_name=event_name, target=target, group=group, args=args, kwargs=kwargs)

    def status(self, i):  # placeholder
        print('dummy status #', i)


def threader(f):
    """ a threading decorator; put @threaded above the callable that needs threading """
    def threading_function(*args, **kwargs):
        threading.Thread(name=f.__name__, target=f, args=args, kwargs=kwargs).start()

    return threading_function


@threader
def t0():
    time.sleep(3)
    my_thread.stop()
    my_thread.join()


class MyTimer:

    def __init__(self, name, kwargs):
        for key, value in kwargs.items():  # unpack kwargs and convert to variables
            setattr(self, key, value)

    def __call__(self, stop_event):
        i = 5
        while not stop_event.is_set() and i > 0:
            print('I: ', i)
            time.sleep(1)
            i -= 1
        if stop_event.is_set():
            print('timer is CANCELLED')
        else:
            print('timer is FINISHED')


def main():
    global t1
    global my_thread
    name = 'timer'
    t1_kwargs = {'interval': 99, 'message': 'STARTING COUNTDOWN...'}
    t1 = MyTimer(name, t1_kwargs)
    my_thread = MyThread(name, target=t1)
    my_thread.status(1)
    my_thread.start()
    print(t1.interval)
    t0()
    my_thread.join()
    my_thread.status(2)
    # second countdown
    my_thread.start()
    my_thread.status(3)
    my_thread.join()
    my_thread.status(4)
    print('END')


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('\b\b\r')
        print("PROGRAM TERMINATED AT USER'S REQUEST")
        force_exit()

Tirages:

dummy status # 1
I:  5
99
I:  4
I:  3
timer is CANCELLED
dummy status # 2
I:  5
dummy status # 3
I:  4
I:  3
I:  2
I:  1
timer is FINISHED
dummy status # 4
END
0
Booboo 15 oct. 2020 à 11:56