Skip to content

Instantly share code, notes, and snippets.

@BigRoy
Last active December 28, 2023 20:33
Show Gist options
  • Save BigRoy/1972822065e38f8fae7521078e44eca2 to your computer and use it in GitHub Desktop.
Save BigRoy/1972822065e38f8fae7521078e44eca2 to your computer and use it in GitHub Desktop.
Pyblish debug stepper - pauses between each plug-in process and shows the Context + Instances with their data at that point in time
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()
@BigRoy
Copy link
Author

BigRoy commented Dec 28, 2023

Tree View debugging

image

Here's a very quick and dirty prototype that shows a similar updating view but then as an expandable tree view for the dictionaries:

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

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


TAB = 4* "&nbsp;"
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):
    def __init__(self, *args, **kwargs):
        super(DictChangesModel, self).__init__(*args, **kwargs)
        
        columns = ["Key", "Type", "Value"]
        self.setColumnCount(len(columns))
        for i, label in enumerate(columns):
            self.setHeaderData(i, QtCore.Qt.Horizontal, label)
        
        self._data = {}

    def _update_recursive(self, data, parent, previous_data):
        for key, value in data.items():
            
            # Key
            key_color = NEW_KEY_COLOR if key not in previous_data else KEY_COLOR
            
            item = QtGui.QStandardItem(key)
            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(value).__name__:
                type_color = NEW_VALUE_TYPE_COLOR
                
            type_item = QtGui.QStandardItem(type_str)
            type_item.setData(type_color, QtCore.Qt.ForegroundRole)
            
            # Value
            value_str = str(value)
            if len(value_str) > MAX_VALUE_STR_LEN:
                value_str = value_str[:MAX_VALUE_STR_LEN] + "..."
            
            value_color = VALUE_COLOR
            if key not in previous_data or previous_data[key] != value:
                value_color = NEW_VALUE_COLOR    
                
            value_item = QtGui.QStandardItem(value_str)
            value_item.setData(value_color, QtCore.Qt.ForegroundRole)
            if value:
                # 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, {})
                self._update_recursive(value, parent=item, previous_data=previous_value)
            
            parent.appendRow([item, type_item, value_item])
            
        self._data = data

    def update(self, data):
        parent = self.invisibleRootItem()
        
        # Clear the existing rows
        if parent.rowCount() > 0:
            parent.removeRows(0, parent.rowCount())
        
        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)
        self.view.expandToDepth(0)

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


window = DebugUI()
window.show()

@BigRoy
Copy link
Author

BigRoy commented Dec 28, 2023

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* "&nbsp;"
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()

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment