Skip to content

Events & Logging

karcytics_sdk.plugin.events

Central EventBus for Karcytics SDK.

Provides a singleton event bus using Qt signals to allow asynchronous and decoupled communication between different plugins and the core application.

karcytics_sdk.plugin.logging

Karcytics SDK Logging Utilities.

Provides Karcytics-aware loggers that automatically attach plugin metadata and route to the central diagnostic engine.

PluginLoggerAdapter

Bases: LoggerAdapter

Logger adapter that injects plugin_id into every log record.

Source code in src/karcytics_sdk/plugin/logging.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class PluginLoggerAdapter(logging.LoggerAdapter):
    """Logger adapter that injects plugin_id into every log record."""

    def __init__(self, logger: logging.Logger, plugin_id: str):
        """Initialize the logger adapter with a specific plugin identifier.

        Args:
            logger: The parent Python logger instance to adapt.
            plugin_id: The unique identifier of the target plugin.
        """
        super().__init__(logger, {"plugin_id": plugin_id})
        self.plugin_id = plugin_id

    def process(self, msg, kwargs):
        """Process log messages to inject the plugin metadata context.

        Args:
            msg: The actual log message content.
            kwargs: Thread or level attributes associated with the log record.

        Returns:
            A tuple of (processed_message, kwargs) containing the updated extra dict.
        """
        # Ensure plugin_id is in the extra dict for the handler to find
        extra = kwargs.get("extra", {})
        extra["plugin_id"] = self.plugin_id
        kwargs["extra"] = extra
        return msg, kwargs

__init__(logger, plugin_id)

Initialize the logger adapter with a specific plugin identifier.

Parameters:

Name Type Description Default
logger Logger

The parent Python logger instance to adapt.

 required
plugin_id str

The unique identifier of the target plugin.

 required
Source code in src/karcytics_sdk/plugin/logging.py 
25
26
27
28
29
30
31
32
33
def __init__(self, logger: logging.Logger, plugin_id: str):
    """Initialize the logger adapter with a specific plugin identifier.

    Args:
        logger: The parent Python logger instance to adapt.
        plugin_id: The unique identifier of the target plugin.
    """
    super().__init__(logger, {"plugin_id": plugin_id})
    self.plugin_id = plugin_id

process(msg, kwargs)

Process log messages to inject the plugin metadata context.

Parameters:

Name Type Description Default
msg

The actual log message content.

 required
kwargs

Thread or level attributes associated with the log record.

 required

Returns:

Type Description

A tuple of (processed_message, kwargs) containing the updated extra dict.
Source code in src/karcytics_sdk/plugin/logging.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def process(self, msg, kwargs):
    """Process log messages to inject the plugin metadata context.

    Args:
        msg: The actual log message content.
        kwargs: Thread or level attributes associated with the log record.

    Returns:
        A tuple of (processed_message, kwargs) containing the updated extra dict.
    """
    # Ensure plugin_id is in the extra dict for the handler to find
    extra = kwargs.get("extra", {})
    extra["plugin_id"] = self.plugin_id
    kwargs["extra"] = extra
    return msg, kwargs

configure_plugin_logging(plugin_id, log_dir=None)

Give an isolated plugin worker its own durable, local log file.

Without this, an isolated worker's root logger has no handlers at all: Python's own last-resort handler only ever prints WARNING+ to stderr, so every .debug()/.info() call a plugin makes is silently dropped, not just unrelayed. PluginDaemon._stderr_reader_loop on the Hub side only tails whatever does reach stderr, and only the last 200 lines, so nothing survives a crash on disk regardless. This adds a rotating file handler under ~/.karcytics/logs/plugin_workers/<plugin_id>.log (kept distinct from the Hub-relayed logs/plugins/<plugin_id>.log path so the two processes never contend as concurrent writers on one file) and a stderr handler so DEBUG+ actually reaches the existing relay too. Idempotent — safe to call more than once for the same process.

Source code in src/karcytics_sdk/plugin/logging.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def configure_plugin_logging(plugin_id: str, log_dir: Path | None = None) -> Path:
    """Give an isolated plugin worker its own durable, local log file.

    Without this, an isolated worker's root logger has no handlers at all:
    Python's own last-resort handler only ever prints WARNING+ to stderr, so
    every `.debug()`/`.info()` call a plugin makes is silently dropped, not
    just unrelayed. `PluginDaemon._stderr_reader_loop` on the Hub side only
    tails whatever *does* reach stderr, and only the last 200 lines, so
    nothing survives a crash on disk regardless. This adds a rotating file
    handler under `~/.karcytics/logs/plugin_workers/<plugin_id>.log` (kept
    distinct from the Hub-relayed `logs/plugins/<plugin_id>.log` path so the
    two processes never contend as concurrent writers on one file) and a
    stderr handler so DEBUG+ actually reaches the existing relay too.
    Idempotent — safe to call more than once for the same process.
    """
    root = logging.getLogger()
    if getattr(root, "_karcytics_plugin_logging_configured", False):
        return root._karcytics_plugin_log_file  # type: ignore[attr-defined]

    if log_dir is None:
        log_dir = Path.home() / ".karcytics" / "logs" / "plugin_workers"
    log_dir.mkdir(parents=True, exist_ok=True)
    log_file = log_dir / f"{plugin_id}.log"

    file_handler = logging.handlers.RotatingFileHandler(
        log_file, maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8"
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s"))

    stream_handler = logging.StreamHandler()
    stream_handler.setLevel(logging.DEBUG)
    stream_handler.setFormatter(logging.Formatter("%(levelname)s %(name)s: %(message)s"))

    root.setLevel(logging.DEBUG)
    for noisy_logger_name in _NOISY_THIRD_PARTY_LOGGERS:
        logging.getLogger(noisy_logger_name).setLevel(logging.WARNING)
    root.addHandler(file_handler)
    root.addHandler(stream_handler)
    root._karcytics_plugin_logging_configured = True  # type: ignore[attr-defined]
    root._karcytics_plugin_log_file = log_file  # type: ignore[attr-defined]
    return log_file

get_logger(name, plugin_id=None)

Get a logger instance, optionally adapted for a specific plugin.

Parameters:

Name Type Description Default
name str

The name of the logger (usually name)

 required
plugin_id str | None

The ID of the plugin this logger belongs to

 None

Returns:

Type Description
Logger | PluginLoggerAdapter

A logging.Logger or PluginLoggerAdapter instance.
Source code in src/karcytics_sdk/plugin/logging.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def get_logger(name: str, plugin_id: str | None = None) -> logging.Logger | PluginLoggerAdapter:
    """Get a logger instance, optionally adapted for a specific plugin.

    Args:
        name: The name of the logger (usually __name__)
        plugin_id: The ID of the plugin this logger belongs to

    Returns:
        A logging.Logger or PluginLoggerAdapter instance.
    """
    logger = logging.getLogger(name)
    if plugin_id:
        return PluginLoggerAdapter(logger, plugin_id)
    return logger

karcytics_sdk.plugin.signals

Standard plugin signals for Karcytics SDK.

This module defines all standard PySide6 signals that plugins should emit to communicate state changes and results to the Karcytics core application.

PluginSignals

Bases: QObject

Standard signals that all plugins emit.

Plugins should use these signals to ensure consistent communication with Karcytics core and other plugins.

Attributes:

Name Type Description
status_message

Emitted with a short status string for the UI status bar
log_message

Emitted with detailed log messages
state_changed

Emitted when plugin state changes
undo_available

Emitted with bool indicating if undo is available
redo_available

Emitted with bool indicating if redo is available
undo_state_changed

Emitted after any change to the undo history — availability, the undo/redo labels, or the clean (saved) state
analysis_started

Emitted when analysis begins
analysis_progress

Emitted with int (0-100) for progress
analysis_complete

Emitted when analysis finishes
analysis_error

Emitted with error message if analysis fails
data_changed

Emitted when underlying data changes
Source code in src/karcytics_sdk/plugin/signals.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class PluginSignals(QObject):
    """Standard signals that all plugins emit.

    Plugins should use these signals to ensure consistent communication
    with Karcytics core and other plugins.

    Attributes:
        status_message: Emitted with a short status string for the UI status bar
        log_message: Emitted with detailed log messages
        state_changed: Emitted when plugin state changes
        undo_available: Emitted with bool indicating if undo is available
        redo_available: Emitted with bool indicating if redo is available
        undo_state_changed: Emitted after any change to the undo history —
            availability, the undo/redo labels, or the clean (saved) state
        analysis_started: Emitted when analysis begins
        analysis_progress: Emitted with int (0-100) for progress
        analysis_complete: Emitted when analysis finishes
        analysis_error: Emitted with error message if analysis fails
        data_changed: Emitted when underlying data changes
    """

    # Status / Logging
    status_message = Signal(str)  # Short status for UI (e.g. "Processing...")
    log_message = Signal(str)  # Detailed log message

    # State Management
    state_changed = Signal()  # Plugin state changed
    undo_available = Signal(bool)  # Whether undo is available
    redo_available = Signal(bool)  # Whether redo is available
    undo_state_changed = Signal()  # Undo labels / availability / clean state changed

    # Analysis Results
    analysis_started = Signal()
    analysis_progress = Signal(int)  # 0-100
    analysis_complete = Signal()
    analysis_error = Signal(str)  # Error message

    # Data Changed
    data_changed = Signal()  # Some data changed (image, parameters, etc)

move_to_gui_thread(obj)

Give a newly built, parentless obj to the application's GUI thread.

Call it right after constructing a long-lived QObject that may be built off the GUI thread (a lazy singleton first touched by a background thread, or anything created from a plain threading.Thread). Under PySide6, a slot that is not a bound QObject method (a lambda, functools.partial) is delivered through the sender's thread; if that thread has no running event loop, every such delivery is silently dropped. Does nothing when no application exists yet or obj is already on the GUI thread.

Source code in src/karcytics_sdk/plugin/signals.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def move_to_gui_thread(obj: QObject) -> None:
    """Give a newly built, parentless `obj` to the application's GUI thread.

    Call it right after constructing a long-lived `QObject` that may be built
    off the GUI thread (a lazy singleton first touched by a background
    thread, or anything created from a plain `threading.Thread`). Under
    PySide6, a slot that is not a bound `QObject` method (a lambda,
    `functools.partial`) is delivered through the sender's thread; if that
    thread has no running event loop, every such delivery is silently
    dropped. Does nothing when no application exists yet or `obj` is already
    on the GUI thread.
    """
    from PySide6.QtCore import QCoreApplication

    app = QCoreApplication.instance()
    if app is not None and obj.thread() is not app.thread():
        obj.moveToThread(app.thread())

safe_disconnect(signal, slot=None)

Disconnect slot from signal, quietly doing nothing if it was not connected.

For the "disconnect, then connect" idiom that keeps a slot connected exactly once. PySide6 reports a disconnect that finds nothing with a RuntimeWarning (and, for a deleted sender, a RuntimeError); neither is useful to a caller that only wants the slot gone.

Parameters:

Name Type Description Default
signal Any

The bound signal, e.g. button.clicked.

 required
slot Callable[..., Any] | None

The slot to remove. None removes every connection.

 None

Returns:

Type Description
bool

True if a connection was removed.
Source code in src/karcytics_sdk/plugin/signals.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def safe_disconnect(signal: Any, slot: Callable[..., Any] | None = None) -> bool:
    """Disconnect `slot` from `signal`, quietly doing nothing if it was not connected.

    For the "disconnect, then connect" idiom that keeps a slot connected
    exactly once. PySide6 reports a disconnect that finds nothing with a
    `RuntimeWarning` (and, for a deleted sender, a `RuntimeError`); neither
    is useful to a caller that only wants the slot gone.

    Args:
        signal: The bound signal, e.g. `button.clicked`.
        slot: The slot to remove. `None` removes every connection.

    Returns:
        True if a connection was removed.
    """
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", RuntimeWarning)
        try:
            return bool(signal.disconnect(slot) if slot is not None else signal.disconnect())
        except (TypeError, RuntimeError):
            return False