从 main() 访问 Python class 实例属性失败
Access to Python class instance attributes from main() fails
我正在编写一个控制程序，它需要多个可配置、可重启的计时器，这些计时器 运行 在它们自己的线程中。最初我用所有计时器的自定义函数和自定义事件名称实现了这一点。因为这是一种非常丑陋的方法，所以我正在使用更类似于 OO 的方法进行重写。请考虑下面的最小示例，它完成了我需要它做的所有事情，但有一个例外，我无法找到解决方案：我无法访问所需的 class 实例属性（我希望我是在这里使用正确的术语）。
失败语句在第 74 行，已被注释掉。我怀疑我传递参数和 kwargs 的方式有问题，但 运行 进入了砖墙。你能看出我哪里出了问题并提出解决方案吗？
#!/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()
首先，您正在尝试初始化您的基础 class 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, 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()
打印：
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
我正在编写一个控制程序，它需要多个可配置、可重启的计时器，这些计时器 运行 在它们自己的线程中。最初我用所有计时器的自定义函数和自定义事件名称实现了这一点。因为这是一种非常丑陋的方法，所以我正在使用更类似于 OO 的方法进行重写。请考虑下面的最小示例，它完成了我需要它做的所有事情，但有一个例外，我无法找到解决方案：我无法访问所需的 class 实例属性（我希望我是在这里使用正确的术语）。
失败语句在第 74 行，已被注释掉。我怀疑我传递参数和 kwargs 的方式有问题，但 运行 进入了砖墙。你能看出我哪里出了问题并提出解决方案吗？
#!/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()
首先，您正在尝试初始化您的基础 class 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, 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()
打印：
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