desktop

desktop

Desktop notification implementation using plyer.

Provides cross-platform desktop notifications for Marianne job events. Uses the plyer library for platform-independent notification support.

Phase 5 of Marianne implementation: Missing README features.

Classes

DesktopNotifier

DesktopNotifier(events=None, app_name='Marianne AI Compose', timeout=10)

Desktop notification implementation using plyer.

Provides cross-platform desktop notifications on Windows, macOS, and Linux. Gracefully degrades if plyer is not installed - logs warning but doesn't fail.

Example usage

notifier = DesktopNotifier( events={NotificationEvent.JOB_COMPLETE, NotificationEvent.JOB_FAILED}, app_name="Marianne", ) await notifier.send(context)

Configuration from YAML

notifications: - type: desktop on_events: [job_complete, job_failed] config: timeout: 10 app_name: "Marianne AI"

Initialize the desktop notifier.

Parameters:

Name	Type	Description	Default
events	set[NotificationEvent] | None	Set of events to subscribe to. Defaults to job-level events.	None
app_name	str	Application name shown in notifications.	'Marianne AI Compose'
timeout	int	Notification display timeout in seconds (platform-dependent).	10

Source code in src/marianne/notifications/desktop.py

def __init__(
    self,
    events: set[NotificationEvent] | None = None,
    app_name: str = "Marianne AI Compose",
    timeout: int = 10,
) -> None:
    """Initialize the desktop notifier.

    Args:
        events: Set of events to subscribe to. Defaults to job-level events.
        app_name: Application name shown in notifications.
        timeout: Notification display timeout in seconds (platform-dependent).
    """
    self._events = events or {
        NotificationEvent.JOB_COMPLETE,
        NotificationEvent.JOB_FAILED,
        NotificationEvent.JOB_PAUSED,
    }
    self._app_name = app_name
    self._timeout = timeout
    self._warned_unavailable = False

Attributes

subscribed_events property

subscribed_events

Events this notifier is registered to receive.

Returns:

Type	Description
set[NotificationEvent]	Set of subscribed NotificationEvent types.

Functions

from_config classmethod

from_config(on_events, config=None)

Create DesktopNotifier from YAML configuration.

Parameters:

Name	Type	Description	Default
on_events	list[str]	List of event name strings from config.	required
config	dict[str, Any] | None	Optional dict with 'app_name' and 'timeout'.	None

Returns:

Type	Description
DesktopNotifier	Configured DesktopNotifier instance.

Example

notifier = DesktopNotifier.from_config( on_events=["job_complete", "job_failed"], config={"timeout": 5, "app_name": "My App"}, )

Source code in src/marianne/notifications/desktop.py

@classmethod
def from_config(
    cls,
    on_events: list[str],
    config: dict[str, Any] | None = None,
) -> "DesktopNotifier":
    """Create DesktopNotifier from YAML configuration.

    Args:
        on_events: List of event name strings from config.
        config: Optional dict with 'app_name' and 'timeout'.

    Returns:
        Configured DesktopNotifier instance.

    Example:
        notifier = DesktopNotifier.from_config(
            on_events=["job_complete", "job_failed"],
            config={"timeout": 5, "app_name": "My App"},
        )
    """
    config = config or {}

    # Convert string event names to NotificationEvent enums
    events: set[NotificationEvent] = set()
    for event_name in on_events:
        try:
            # Handle both "job_complete" and "JOB_COMPLETE" formats
            normalized = event_name.upper()
            events.add(NotificationEvent[normalized])
        except KeyError:
            _logger.warning("unknown_notification_event", event_name=event_name)

    return cls(
        events=events if events else None,
        app_name=config.get("app_name", "Marianne AI Compose"),
        timeout=config.get("timeout", 10),
    )

send async

send(context)

Send a desktop notification.

Uses plyer for cross-platform notification support. Fails gracefully if plyer is unavailable or notification fails.

Parameters:

Name	Type	Description	Default
context	NotificationContext	Notification context with event details.	required

Returns:

Type	Description
bool	True if notification was sent, False if unavailable or failed.

Source code in src/marianne/notifications/desktop.py

async def send(self, context: NotificationContext) -> bool:
    """Send a desktop notification.

    Uses plyer for cross-platform notification support.
    Fails gracefully if plyer is unavailable or notification fails.

    Args:
        context: Notification context with event details.

    Returns:
        True if notification was sent, False if unavailable or failed.
    """
    if not _PLYER_AVAILABLE:
        if not self._warned_unavailable:
            _logger.warning(
                "Desktop notifications unavailable - install plyer: "
                "pip install plyer"
            )
            self._warned_unavailable = True
        return False

    if context.event not in self._events:
        # Not subscribed to this event
        return True

    title = context.format_title()
    message = context.format_message()

    try:
        # plyer.notification.notify is synchronous, but we wrap it
        # in the async interface for consistency
        _notification_module.notify(
            title=title,
            message=message,
            app_name=self._app_name,
            timeout=self._timeout,
        )
        _logger.debug("desktop_notification_sent", title=title)
        return True

    except Exception as e:
        # Various platform-specific errors can occur
        # (missing system notification service, etc.)
        _logger.warning("desktop_notification_failed", error=str(e))
        return False

close async

close()

Clean up resources.

Desktop notifier has no resources to clean up, but implements the protocol method.

Source code in src/marianne/notifications/desktop.py

async def close(self) -> None:
    """Clean up resources.

    Desktop notifier has no resources to clean up,
    but implements the protocol method.
    """
    pass

MockDesktopNotifier

MockDesktopNotifier(events=None)

Mock desktop notifier for testing.

Records all notifications sent without actually displaying them. Useful for testing notification integration.

Initialize mock notifier.

Parameters:

Name	Type	Description	Default
events	set[NotificationEvent] | None	Set of events to subscribe to.	None

Source code in src/marianne/notifications/desktop.py

def __init__(
    self,
    events: set[NotificationEvent] | None = None,
) -> None:
    """Initialize mock notifier.

    Args:
        events: Set of events to subscribe to.
    """
    self._events = events or {
        NotificationEvent.JOB_COMPLETE,
        NotificationEvent.JOB_FAILED,
    }
    self.sent_notifications: list[NotificationContext] = []
    self._fail_next = False

Attributes

subscribed_events property

subscribed_events

Events this notifier handles.

Functions

set_fail_next

set_fail_next(should_fail=True)

Configure the next send() call to fail.

Parameters:

Name	Type	Description	Default
should_fail	bool	If True, next send() returns False.	True

Source code in src/marianne/notifications/desktop.py

def set_fail_next(self, should_fail: bool = True) -> None:
    """Configure the next send() call to fail.

    Args:
        should_fail: If True, next send() returns False.
    """
    self._fail_next = should_fail

send async

send(context)

Record notification without displaying.

Parameters:

Name	Type	Description	Default
context	NotificationContext	Notification context.	required

Returns:

Type	Description
bool	True unless set_fail_next was called.

Source code in src/marianne/notifications/desktop.py

async def send(self, context: NotificationContext) -> bool:
    """Record notification without displaying.

    Args:
        context: Notification context.

    Returns:
        True unless set_fail_next was called.
    """
    if self._fail_next:
        self._fail_next = False
        return False

    self.sent_notifications.append(context)
    return True

close async

close()

Clear recorded notifications.

Source code in src/marianne/notifications/desktop.py

async def close(self) -> None:
    """Clear recorded notifications."""
    self.sent_notifications.clear()

get_notification_count

get_notification_count()

Get number of recorded notifications.

Source code in src/marianne/notifications/desktop.py

def get_notification_count(self) -> int:
    """Get number of recorded notifications."""
    return len(self.sent_notifications)

get_notifications_for_event

get_notifications_for_event(event)

Get all notifications for a specific event type.

Parameters:

Name	Type	Description	Default
event	NotificationEvent	Event type to filter by.	required

Returns:

Type	Description
list[NotificationContext]	List of matching notification contexts.

Source code in src/marianne/notifications/desktop.py

def get_notifications_for_event(
    self, event: NotificationEvent
) -> list[NotificationContext]:
    """Get all notifications for a specific event type.

    Args:
        event: Event type to filter by.

    Returns:
        List of matching notification contexts.
    """
    return [n for n in self.sent_notifications if n.event == event]

Functions

is_desktop_notification_available

is_desktop_notification_available()

Check if desktop notifications are available.

Returns:

Type	Description
bool	True if plyer is installed and can send notifications.

Source code in src/marianne/notifications/desktop.py

def is_desktop_notification_available() -> bool:
    """Check if desktop notifications are available.

    Returns:
        True if plyer is installed and can send notifications.
    """
    return _PLYER_AVAILABLE