Pyblish debug stepper - pauses between each plug-in process and shows the Context + Instances with their data at that point in time

pyblish_debug_stepper.py

import pprint
import inspect
import html
import copy

import pyblish.api
from Qt import QtWidgets, QtCore, QtGui


TAB = 4* " "
HEADER_SIZE = "15px"
KEY_COLOR = "#ffffff"
NEW_KEY_COLOR = "#00ff00"
VALUE_TYPE_COLOR = "#ffbbbb"
VALUE_COLOR = "#777799"
NEW_VALUE_COLOR = "#DDDDCC"

COLORED = "<font style='color:{color}'>{txt}</font>"

MAX_VALUE_STR_LEN = 100

def format_data(data, previous_data):
    previous_data = previous_data or {}
    msg = ""
    for key, value in sorted(data.items()):
        type_str = type(value).__name__
        
        key_color = NEW_KEY_COLOR if key not in previous_data else KEY_COLOR
        value_color = VALUE_COLOR
        if key not in previous_data or previous_data[key] != value:
            value_color = NEW_VALUE_COLOR
        
        value_str = str(value)
        if len(value_str) > MAX_VALUE_STR_LEN:
            value_str = value_str[:MAX_VALUE_STR_LEN] + "..."
            
        key_str = COLORED.format(txt=key, color=key_color)
        type_str = COLORED.format(txt=type_str, color=VALUE_TYPE_COLOR)
        value_str = COLORED.format(txt=html.escape(value_str), color=value_color)
        
        data_str = TAB + f"{key_str} ({type_str}): {value_str} <br>"
        
        msg += data_str
    return msg
    

class DebugUI(QtWidgets.QDialog):

    def __init__(self, parent=None):
        super(DebugUI, self).__init__(parent=parent)
        
        self.setWindowTitle("Pyblish Debug Stepper")
        self.setWindowFlags(
            QtCore.Qt.Window
            | QtCore.Qt.CustomizeWindowHint
            | QtCore.Qt.WindowTitleHint
            | QtCore.Qt.WindowMinimizeButtonHint
            | QtCore.Qt.WindowCloseButtonHint
            | QtCore.Qt.WindowStaysOnTopHint
        )
        
        layout = QtWidgets.QVBoxLayout(self)
        text_edit = QtWidgets.QTextEdit()
        font = QtGui.QFont("NONEXISTENTFONT")
        font.setStyleHint(font.TypeWriter)
        text_edit.setFont(font)
        text_edit.setLineWrapMode(text_edit.NoWrap)

        step = QtWidgets.QPushButton("Step")
        step.setEnabled(False)

        layout.addWidget(text_edit)
        layout.addWidget(step)

        step.clicked.connect(self.on_step)

        self._pause = False
        self.text = text_edit
        self.step = step
        self.resize(700, 500)
        
        self._previous_data = {}

    def pause(self, state):
        self._pause = state
        self.step.setEnabled(state)

    def on_step(self):
        self.pause(False)

    def showEvent(self, event):
        print("Registering callback..")
        pyblish.api.register_callback("pluginProcessed",
                                      self.on_plugin_processed)

    def hideEvent(self, event):
        self.pause(False)
        print("Deregistering callback..")
        pyblish.api.deregister_callback("pluginProcessed",
                                        self.on_plugin_processed)

    def on_plugin_processed(self, result):
        self.pause(True)
        
        # Don't tell me why - but the pyblish event does not
        # pass along the context with the result. And thus
        # it's non trivial to debug step by step. So, we
        # get the context like the evil bastards we are.
        i = 0
        found_context = None
        current_frame = inspect.currentframe()
        for frame_info in inspect.getouterframes(current_frame):
            frame_locals = frame_info.frame.f_locals
            if "context" in frame_locals:
                found_context = frame_locals["context"]
                break
            i += 1
            if i > 5:
                print("Warning: Pyblish context not found..")
                # We should be getting to the context within 
                # a few frames
                break
        
        plugin_name = result["plugin"].__name__
        duration = result['duration']
        plugin_instance = result["instance"]
        
        msg = ""
        msg += f"Plugin: {plugin_name}"
        if plugin_instance is not None:
            msg += f" -> instance: {plugin_instance}"
        msg += "<br>"
        msg += f"Duration: {duration}ms<br>"
        msg += "====<br>"
        
        context = found_context
        if context is not None:
            id = "context"
            
            msg += f"""<font style='font-size: {HEADER_SIZE};'><b>Context:</b></font><br>"""
            msg += format_data(context.data, previous_data=self._previous_data.get(id))
            msg += "====<br>"
            self._previous_data[id] = copy.deepcopy(context.data)
            
            for instance in context:
                id = instance.name
                
                msg += f"""<font style='font-size: {HEADER_SIZE};'><b>Instance:</b> {instance}</font><br>"""
                msg += format_data(instance.data, previous_data=self._previous_data.get(id))
                msg += "----<br>"
                self._previous_data[id] = copy.deepcopy(instance.data)
        
        self.text.setHtml(msg)

        app = QtWidgets.QApplication.instance()
        while self._pause:
            # Allow user interaction with the UI
            app.processEvents()


window = DebugUI()
window.show()

Note that this example uses a customized pluginProcessedContext event that Pyblish does NOT emit by default

And quick and dirty with updating existing items instead of clearing and always adding new:

import pprint
import inspect
import contextlib
import html
import copy
import json

import pyblish.api
from Qt import QtWidgets, QtCore, QtGui


TAB = 4* " "
HEADER_SIZE = "15px"

KEY_COLOR = QtGui.QColor("#ffffff")
NEW_KEY_COLOR = QtGui.QColor("#00ff00")
VALUE_TYPE_COLOR = QtGui.QColor("#ffbbbb")
NEW_VALUE_TYPE_COLOR = QtGui.QColor("#ff4444")
VALUE_COLOR = QtGui.QColor("#777799")
NEW_VALUE_COLOR = QtGui.QColor("#DDDDCC")
CHANGED_VALUE_COLOR = QtGui.QColor("#CCFFCC")


MAX_VALUE_STR_LEN = 100
    
    
def failsafe_deepcopy(data):
    """Allow skipping the deepcopy for unsupported types"""
    try:
        return copy.deepcopy(data)
    except TypeError:
        if isinstance(data, dict):
            return {
                key: failsafe_deepcopy(value)
                for key, value in data.items()
            }
        elif isinstance(data, list):
            return data.copy()
    return data
    

class DictChangesModel(QtGui.QStandardItemModel):
    # TODO: Replace this with a QAbstractItemModel
    def __init__(self, *args, **kwargs):
        super(DictChangesModel, self).__init__(*args, **kwargs)
        self._data = {}
        
        columns = ["Key", "Type", "Value"]
        self.setColumnCount(len(columns))
        for i, label in enumerate(columns):
            self.setHeaderData(i, QtCore.Qt.Horizontal, label)

    def _update_recursive(self, data, parent, previous_data):
        for key, value in data.items():
            
            # Find existing item or add new row
            parent_index = parent.index()
            for row in range(self.rowCount(parent_index)):
                # Update existing item if it exists
                index = self.index(row, 0, parent_index)
                if index.data() == key:
                    item = self.itemFromIndex(index)
                    type_item = self.itemFromIndex(self.index(row, 1, parent_index))
                    value_item = self.itemFromIndex(self.index(row, 2, parent_index))
                    break
            else:
                item = QtGui.QStandardItem(key)
                type_item = QtGui.QStandardItem()
                value_item = QtGui.QStandardItem()
                parent.appendRow([item, type_item, value_item])
            
            # Key
            key_color = NEW_KEY_COLOR if key not in previous_data else KEY_COLOR
            item.setData(key_color, QtCore.Qt.ForegroundRole)
            
            # Type
            type_str = type(value).__name__
            type_color = VALUE_TYPE_COLOR
            if key in previous_data and type(previous_data[key]).__name__ != type_str:
                type_color = NEW_VALUE_TYPE_COLOR
                
            type_item.setText(type_str)
            type_item.setData(type_color, QtCore.Qt.ForegroundRole)
            
            # Value
            value_changed = False
            if key not in previous_data or previous_data[key] != value:
                value_changed = True
            value_color = NEW_VALUE_COLOR if value_changed else VALUE_COLOR
                
            value_item.setData(value_color, QtCore.Qt.ForegroundRole)
            if value_changed:
                value_str = str(value)
                if len(value_str) > MAX_VALUE_STR_LEN:
                    value_str = value_str[:MAX_VALUE_STR_LEN] + "..."
                value_item.setText(value_str)
                # Preferably this is deferred to only when the data gets requested
                # since this formatting can be slow for very large data sets like
                # project settings and system settings
                # This will also be MUCH MUCH faster if we don't clear the items on each update
                # but only updated/add/remove changed items so that this also runs much less often
                value_item.setData(json.dumps(value, default=str, indent=4), QtCore.Qt.ToolTipRole)
            
            
            if isinstance(value, dict):
                previous_value = previous_data.get(key, {})
                if previous_data.get(key) != value:
                    # Update children if the value is not the same as before
                    self._update_recursive(value, parent=item, previous_data=previous_value)
                else:
                    # TODO: Ensure all children are updated to be not marked as 'changed'
                    #   in the most optimal way possible 
                    self._update_recursive(value, parent=item, previous_data=previous_value)
                
        self._data = data

    def update(self, data):
        parent = self.invisibleRootItem()
        
        data = failsafe_deepcopy(data)
        previous_data = self._data
        self._update_recursive(data, parent, previous_data)
        self._data = data  # store previous data for next update
        
            

class DebugUI(QtWidgets.QDialog):

    def __init__(self, parent=None):
        super(DebugUI, self).__init__(parent=parent)
        
        self._set_window_title()
        self.setWindowFlags(
            QtCore.Qt.Window
            | QtCore.Qt.CustomizeWindowHint
            | QtCore.Qt.WindowTitleHint
            | QtCore.Qt.WindowMinimizeButtonHint
            | QtCore.Qt.WindowCloseButtonHint
            | QtCore.Qt.WindowStaysOnTopHint
        )
        
        layout = QtWidgets.QVBoxLayout(self)
        text_edit = QtWidgets.QTextEdit()
        text_edit.setFixedHeight(65)
        font = QtGui.QFont("NONEXISTENTFONT")
        font.setStyleHint(font.TypeWriter)
        text_edit.setFont(font)
        text_edit.setLineWrapMode(text_edit.NoWrap)

        step = QtWidgets.QPushButton("Step")
        step.setEnabled(False)
        
        model = DictChangesModel()
        proxy = QtCore.QSortFilterProxyModel()
        proxy.setSourceModel(model)
        view = QtWidgets.QTreeView()
        view.setModel(proxy)
        view.setSortingEnabled(True)

        layout.addWidget(text_edit)
        layout.addWidget(view)
        layout.addWidget(step)

        step.clicked.connect(self.on_step)

        self._pause = False
        self.model = model
        self.proxy = proxy
        self.view = view
        self.text = text_edit
        self.step = step
        self.resize(700, 500)
        
        self._previous_data = {}
        
    def _set_window_title(self, plugin=None):
        title = "Pyblish Debug Stepper"
        if plugin is not None:
            plugin_label = plugin.label or plugin.__name__
            title += f" | {plugin_label}"
        self.setWindowTitle(title)

    def pause(self, state):
        self._pause = state
        self.step.setEnabled(state)

    def on_step(self):
        self.pause(False)

    def showEvent(self, event):
        print("Registering callback..")
        pyblish.api.register_callback("pluginProcessedContext",
                                      self.on_plugin_processed)

    def hideEvent(self, event):
        self.pause(False)
        print("Deregistering callback..")
        pyblish.api.deregister_callback("pluginProcessedContext",
                                        self.on_plugin_processed)

    def on_plugin_processed(self, context, result):
        self.pause(True)
        
        self._set_window_title(plugin=result["plugin"])
        
        plugin_order = result["plugin"].order
        plugin_name = result["plugin"].__name__
        duration = result['duration']
        plugin_instance = result["instance"]
        
        msg = ""
        msg += f"Order: {plugin_order}<br>"
        msg += f"Plugin: {plugin_name}"
        if plugin_instance is not None:
            msg += f" -> instance: {plugin_instance}"
        msg += "<br>"
        msg += f"Duration: {duration} ms<br>"
        self.text.setHtml(msg)
        
        data = {
            "context": context.data
        }
        for instance in context:
            data[instance.name] = instance.data
        self.model.update(data)

        app = QtWidgets.QApplication.instance()
        while self._pause:
            # Allow user interaction with the UI
            app.processEvents()


window = DebugUI()
window.show()

A Tip for whom it may concern,
There's a ready to run pyblish_debug_stepper for AYON here
where, you can copy & paste it to your script editor in your DCC and then run

window = DebugUI()
window.show()

---

Alternatively, You can get the tool by checking out my PR Feature/add pyblish debug stepper to experimental tools #753
Where I've included the Pyblish Debug Stepper as an experimental tool that can be accessed from any Host/DCC.

For more info, please refer to Pyblish Plugins Debugging | Ynput Forums

---

Many thanks for BigRoy for making this awesome tool.