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