问题描述
我正在编写一个控制程序，该程序需要多个可配置的，可重新启动的计时器，它们在各自的线程中运行。最初，我使用所有计时器的自定义函数和自定义事件名称来实现此功能。因为这是一个非常丑陋的方法，所以我正在使用一种更类似于OO的方法进行重写。请考虑下面的最小示例，该示例可以完成我需要做的所有事情，但有一个我无法找到解决方案的例外：我无法访问所需的类实例属性（我希望我使用正确的术语）这里）。
失败的语句在第74行并已注释掉。我怀疑传递参数和扭曲的方式存在问题，但遇到了障碍。您能看到我出了什么问题并提出解决方案吗？
#!/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,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()
解决方法
首先，您尝试两次初始化基类[{ 'id':2,'name': 'AB','subject': ['Physics','Engineering'],'type':'Permanent','Location':'NY'}]，首先使用
Thread，然后加上：
Thread.__init__(self)
您应该摆脱第一次初始化。
您的问题是您正在为super().__init__(name=name,event_name=event_name,target=target,group=group,args=args,kwargs=kwargs)
实例分配t1：
MyThread但是t1 = MyThread(name=name,target=MyTimer(name,t1_kwargs))
是您的interval实例的属性，而不是您的MyTimer实例的属性。因此MyThread将失败。您想要：
t1.interval您只需要保持两个不同的实例不变即可：
t1 = MyTimer(name,t1_kwargs)
my_thread = MyThread(name,target=t1)
my_thread.status(1)
my_thread.start()
etc.
打印：
#!/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,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,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()