Skip to content

Instantly share code, notes, and snippets.

@papr
Last active March 27, 2019 13:17
Show Gist options
  • Save papr/0eaba2bc70755ffb1ad05385a3458a07 to your computer and use it in GitHub Desktop.
Save papr/0eaba2bc70755ffb1ad05385a3458a07 to your computer and use it in GitHub Desktop.
import logging
import time
from pyglui import ui
from plugin import Plugin
from background_helper import IPC_Logging_Task_Proxy
logger = logging.getLogger(__name__)
def simple_generator(start_value, stride=1, sleep_duration=0.5):
try:
logger.info("Generator start")
while True:
yield start_value
start_value += stride
time.sleep(sleep_duration)
except GeneratorExit:
# Called if the process is being cancelled by teh foreground.
# Do any cleanup here that is required before shutting down.
logger.info("Generator exit")
class Online_Task_Proxy(Plugin):
icon_chr = "TP"
def __init__(self, g_pool):
super().__init__(g_pool)
self.current_foreground_value = float("nan")
self.proxy = None
def cleanup(self):
self.shutdown_background_task()
def recent_events(self, events):
if self.proxy and not self.proxy.canceled:
for recent_value in self.proxy.fetch():
self.current_foreground_value = recent_value
def restart_background_task(self):
self.shutdown_background_task()
self.start_background_task()
def start_background_task(self):
start_value = 42
stride = 3
args = (start_value,)
kwargs = {"stride": stride}
self.proxy = IPC_Logging_Task_Proxy(
"Example Background Task Name", simple_generator, args=args, kwargs=kwargs
)
def shutdown_background_task(self):
if self.proxy:
self.proxy.cancel()
self.proxy = None
def init_ui(self):
self.add_menu()
self.menu.label = "Online Task Proxy"
self.menu.append(
ui.Text_Input(
"current_foreground_value",
self,
label="Current Value",
setter=lambda x: None,
)
)
self.menu.append(
ui.Button("Restart Background Task", self.restart_background_task)
)
def deinit_ui(self):
self.remove_menu()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment