online_task_proxy.py

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()