+#!/usr/bin/env python3
+"""
+ On-Screen Display Python Client Library
+ (c) 2022 Jiri Kalvoda <jirikalvoda@kam.mff.cuni.cz>
+
+
+Basic usage:
+============
+
+osd.notify("first line", "second line")
+
+Each argument is either a single string (a notification line)
+or a pair of strings ("property", "value"). For available
+properties consult OSD Daemon documentation. These arguments
+are position sensitive.
+
+Properties may also be added as named parameters. For example:
+ osd.notify("Hello world", color="red")
+
+Due to the notation of OSD parameters and Python limitations of named
+arguments, all occurrences of '_' in the argument name are be replaced by '-'.
+
+Named arguments are treated as positional arguments placed before all others,
+in unspecified relative order. This behavior was chosen as in most cases, these
+arguments will affect all notification lines. If you need more control, use the
+positional variant.
+
+Therefore, these two invocations are equivalent:
+ osd.notify("a line", outline_color="red")
+ osd.notify(("outline-color", "red"), "a line")
+
+Connection
+==========
+
+If you don't want to send the notification to the default display,
+you can create new connection class via `osd.new_connection(DISPLAY_NAME)`
+or using `osd.Display` constructor which takes an instance of `Xlib.display.Display`.
+
+The instance of `Connection` class has the same `notify` method.
+
+You can reassign a default connection into `osd.default_connection`.
+
+Errors
+======
+
+If there is a problem with connection to the X server, library raises the corresponding
+Xlib exceptions.
+
+If a message contains any forbidden characters such as new line,
+OSDArgumentError will be raised. The `notify_sanitized` method
+filters out forbidden characters, therefore it never raises OSDArgumentError.
+"""
+
+import Xlib
+from Xlib.display import Display
+import Xlib.X
+from Xlib.xobject.drawable import Window
+
+from typing import Callable, List, Tuple, Optional, Union
+
+
+class OSDArgumentError(RuntimeError):
+ pass
+
+
+class Connection():
+ """ See module documentation. """
+ display: Display
+ root_window: Window
+ osd_queue_atom: int
+
+ def __init__(self, display: Display):
+ self.display = display
+ self.root_window = self.display.screen().root
+ self.osd_queue_atom = self.display.get_atom('OSD_QUEUE')
+
+ def _send_raw(self, val: str) -> None:
+ self.root_window.change_property(
+ self.osd_queue_atom,
+ property_type=Xlib.Xatom.STRING,
+ format=8,
+ data=val.encode(),
+ mode=Xlib.X.PropModeAppend
+ )
+ self.display.flush()
+
+ def notify_with_error_handler(self, error_handler: Callable[[str, str, bool], str], *args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
+ lines: List[str] = []
+
+ def add_line(key, val):
+ def remove_char(s, ch, is_key):
+ out = []
+ for i in s:
+ if i == ch:
+ out.append(error_handler(s, ch, is_key))
+ else:
+ out.append(i)
+ return "".join(out)
+
+ key = remove_char(remove_char(str(key), "\n", True), ":", True)
+ val = remove_char(str(val), "\n", False)
+
+ lines.append(f"{key}:{val}")
+
+ for key, val in kwargs.items():
+ add_line(key.replace("_", "-"), val)
+ for x in args:
+ if isinstance(x, tuple):
+ key, val = x
+ add_line(key, val)
+ else:
+ add_line("", x)
+ msg = "\n".join(lines) + "\n\n"
+ self._send_raw(msg)
+
+ def notify(self, *args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
+ def error_handler(s, ch, is_key):
+ raise OSDArgumentError(f"{'Key' if is_key else 'Value'} {repr(s)} contain forbidden character {repr(ch)}.")
+ self.notify_with_error_handler(error_handler, *args, **kwargs)
+
+ def notify_sanitized(self, *args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
+ def error_handler(s, ch, is_key):
+ return ""
+ self.notify_with_error_handler(error_handler, *args, **kwargs)
+
+
+def new_connection(display_name: Optional[str] = None) -> Connection:
+ """ See module documentation. """
+ return Connection(Display(display_name))
+
+
+default_connection: Optional[Connection] = None
+
+
+def _default_or_new_connection() -> Connection:
+ global default_connection
+ if default_connection is None:
+ default_connection = new_connection()
+ return default_connection
+
+
+def notify(*args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
+ """ See module documentation. """
+ _default_or_new_connection().notify(*args, **kwargs)
+
+
+def notify_sanitized(*args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
+ """ See module documentation. """
+ _default_or_new_connection().notify_sanitized(*args, **kwargs)
projects / osdd.git / commitdiff