blob: 14610808665adf73e761b9004a9fa38aaebb94b2 [file] [log] [blame]
# Copyright 2021 The Pigweed Authors
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Window pane base class."""
import asyncio
import logging
from threading import Thread
import time
from typing import Callable, Optional
from pw_console.get_pw_console_app import get_pw_console_app
class PluginMixin:
"""Handles background task management in a Pigweed Console plugin.
Pigweed Console plugins can inherit from this class if they require running
tasks in the background. This is important as any plugin code not in its
own dedicated thread can potentially block the user interface
Example usage: ::
import logging
from pw_console.plugin_mixin import PluginMixin
from pw_console.widgets import WindowPaneToolbar
class AwesomeToolbar(WindowPaneToolbar, PluginMixin):
def __init__(self, *args, **kwargs):
# Call parent class WindowPaneToolbar.__init__
super().__init__(*args, **kwargs)
# Set PluginMixin to execute
# self._awesome_background_task every 10 seconds.
# This function will be run in a separate thread every 10 seconds.
def _awesome_background_task(self) -> bool:
time.sleep(1) # Do real work here.
if self.new_data_processed:
# If new data was processed, and the user interface
# should be updated return True.
# Log using self.plugin_logger for debugging.
self.plugin_logger.debug('New data processed')
# Return True to signal a UI redraw.
return True
# Returning False means no updates needed.
return False
plugin_callback: Callable that is run in a background thread.
plugin_callback_frequency: Number of seconds to wait between
executing plugin_callback.
plugin_logger: logging instance for this plugin. Useful for debugging
code running in a separate thread.
plugin_callback_future: `Future`_ object for the plugin background task.
plugin_event_loop: asyncio event loop running in the background thread.
plugin_enable_background_task: If True, keep periodically running
plugin_callback at the desired frequency. If False the background
task will stop.
.. _Future:
def plugin_init(
plugin_callback: Optional[Callable[..., bool]] = None,
plugin_callback_frequency: float = 30.0,
plugin_logger_name: Optional[str] = 'pw_console_plugins',
) -> None:
"""Call this on __init__() to set plugin background task variables.
plugin_callback: Callable to run in a separate thread from the
Pigweed Console UI. This function should return True if the UI
should be redrawn after execution.
plugin_callback_frequency: Number of seconds to wait between
executing plugin_callback.
plugin_logger_name: Unique name for this plugin's Python
logger. Useful for debugging code running in a separate thread.
self.plugin_callback = plugin_callback
self.plugin_callback_frequency = plugin_callback_frequency
self.plugin_logger = logging.getLogger(plugin_logger_name)
self.plugin_callback_future = None
# Event loop for executing plugin code.
self.plugin_event_loop = asyncio.new_event_loop()
self.plugin_enable_background_task = True
def plugin_start(self):
"""Function used to start this plugin's background thead and task."""
# Create an entry point for the plugin thread.
def _plugin_thread_entry():
# Disable log propagation
self.plugin_logger.propagate = False
# Create a thread for running user code so the UI isn't blocked.
thread = Thread(target=_plugin_thread_entry, args=(), daemon=True)
self.plugin_logger.debug('Starting plugin: %s', self)
if self.plugin_callback is None:
self.plugin_enable_background_task = True
self.plugin_callback_future = asyncio.run_coroutine_threadsafe(
# This function will be executed in a separate thread.
# Using this asyncio event loop.
self.plugin_event_loop) # type: ignore
def plugin_stop(self):
self.plugin_enable_background_task = False
async def _plugin_periodically_run_callback(self) -> None:
while self.plugin_enable_background_task:
start_time = time.time()
# Run the callback and redraw the UI if return value is True
if self.plugin_callback and self.plugin_callback():
run_time = time.time() - start_time
await asyncio.sleep(self.plugin_callback_frequency - run_time)