python-osd/osd/__init__.py

   1 #!/usr/bin/env python3
   2 """
   3                   On-Screen Display Python Client Library
   4              (c) 2022 Jiri Kalvoda <jirikalvoda@kam.mff.cuni.cz>
   5
   6
   7 Basic usage:
   8 ============
   9
  10 import osd
  11 osd.notify("first line", "second line")
  12
  13 Each argument is either a single string (a notification line)
  14 or a pair of strings ("property", "value"). For available
  15 properties consult OSD Daemon documentation. These arguments
  16 are position sensitive.
  17
  18 Properties may also be added as named parameters. For example:
  19     osd.notify("Hello world", color="red")
  20
  21 Due to the notation of OSD parameters and Python limitations of named
  22 arguments, all occurrences of '_' in the argument name are be replaced by '-'.
  23
  24 Named arguments are treated as positional arguments placed before all others,
  25 in unspecified relative order. This behavior was chosen as in most cases, these
  26 arguments will affect all notification lines. If you need more control, use the
  27 positional variant.
  28
  29 Therefore, these two invocations are equivalent:
  30     osd.notify("a line", outline_color="red")
  31     osd.notify(("outline-color", "red"), "a line")
  32
  33 Connection
  34 ==========
  35
  36 If you don't want to send the notification to the default display,
  37 you can create new connection class via `osd.new_connection(DISPLAY_NAME)`
  38 or using `osd.Display` constructor which takes an instance of `Xlib.display.Display`.
  39
  40 The instance of `Connection` class has the same `notify` method.
  41
  42 You can reassign a default connection into `osd.default_connection`.
  43
  44 Errors
  45 ======
  46
  47 If there is a problem with connection to the X server, library raises the corresponding
  48 Xlib exceptions.
  49
  50 If a message contains any forbidden characters such as new line,
  51 OSDArgumentError will be raised. The `notify_sanitized` method
  52 filters out forbidden characters, therefore it never raises OSDArgumentError.
  53 """
  54
  55 import Xlib
  56 from Xlib.display import Display
  57 import Xlib.X
  58 from Xlib.xobject.drawable import Window
  59
  60 from typing import Callable, List, Tuple, Optional, Union
  61
  62
  63 class OSDArgumentError(RuntimeError):
  64     pass
  65
  66
  67 class Connection():
  68     """ See module documentation. """
  69     display: Display
  70     root_window: Window
  71     osd_queue_atom: int
  72
  73     def __init__(self, display: Display):
  74         self.display = display
  75         self.root_window = self.display.screen().root
  76         self.osd_queue_atom = self.display.get_atom('OSD_QUEUE')
  77
  78     def _send_raw(self, val: str) -> None:
  79         self.root_window.change_property(
  80             self.osd_queue_atom,
  81             property_type=Xlib.Xatom.STRING,
  82             format=8,
  83             data=val.encode(),
  84             mode=Xlib.X.PropModeAppend
  85         )
  86         self.display.flush()
  87
  88     def notify_with_error_handler(self, error_handler: Callable[[str, str, bool], str], *args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
  89         lines: List[str] = []
  90
  91         def add_line(key, val):
  92             def remove_char(s, ch, is_key):
  93                 out = []
  94                 for i in s:
  95                     if i == ch:
  96                         out.append(error_handler(s, ch, is_key))
  97                     else:
  98                         out.append(i)
  99                 return "".join(out)
 100
 101             key = remove_char(remove_char(str(key), "\n", True), ":", True)
 102             val = remove_char(str(val), "\n", False)
 103
 104             lines.append(f"{key}:{val}")
 105
 106         for key, val in kwargs.items():
 107             add_line(key.replace("_", "-"), val)
 108         for x in args:
 109             if isinstance(x, tuple):
 110                 key, val = x
 111                 add_line(key, val)
 112             else:
 113                 add_line("", x)
 114         msg = "\n".join(lines) + "\n\n"
 115         self._send_raw(msg)
 116
 117     def notify(self, *args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
 118         def error_handler(s, ch, is_key):
 119             raise OSDArgumentError(f"{'Key' if is_key else 'Value'} {repr(s)} contain forbidden character {repr(ch)}.")
 120         self.notify_with_error_handler(error_handler, *args, **kwargs)
 121
 122     def notify_sanitized(self, *args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
 123         def error_handler(s, ch, is_key):
 124             return ""
 125         self.notify_with_error_handler(error_handler, *args, **kwargs)
 126
 127
 128 def new_connection(display_name: Optional[str] = None) -> Connection:
 129     """ See module documentation. """
 130     return Connection(Display(display_name))
 131
 132
 133 default_connection: Optional[Connection] = None
 134
 135
 136 def _default_or_new_connection() -> Connection:
 137     global default_connection
 138     if default_connection is None:
 139         default_connection = new_connection()
 140     return default_connection
 141
 142
 143 def notify(*args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
 144     """ See module documentation. """
 145     _default_or_new_connection().notify(*args, **kwargs)
 146
 147
 148 def notify_sanitized(*args: Union[str, Tuple[str, str]], **kwargs: str) -> None:
 149     """ See module documentation. """
 150     _default_or_new_connection().notify_sanitized(*args, **kwargs)