not2osd

@@ -24,7 +24,7 @@ import dbus
 import dbus.service
 import dbus.mainloop.glib
 import os
-import subprocess
+import osd
 import sys
 import time
 import re
@@ -75,19 +75,12 @@ class NotificationFetcher(dbus.service.Object):
         #print("expire_timeout", expire_timeout)
         su = strip_tags(summary[0:300]).split("\n")
         bo = strip_tags(body[0:500]).split("\n")
-        for i in range(len(su)):
-            if su[i]!="" and su[i][0]=='-':
-                su[i] = ' '+su[i];
-        for i in range(len(bo)):
-            if bo[i]!="" and bo[i][0]=='-':
-                bo[i] = ' '+bo[i];
         for i in range(len(bo)):
             if bo[i] == "This site has been updated in the background.":
                 return notification_id
         if (expire_timeout <= 0) or (expire_timeout > 20000):
             expire_timeout = 5000
-        cmd = ["osdc","--duration="+str(expire_timeout),"--min-duration=1000","--color=#99F","--outline-color=red"] + su + ["--outline-color=black"] + bo
-        subprocess.call(cmd)
+        osd.notify(("duration", str(expire_timeout)), ("outline_color", "red"), *su, ("outline_color", "black"), *bo, color="#99F", duration=str(expire_timeout), min_duration="1000")
         return notification_id
 
     @dbus.service.method("org.freedesktop.Notifications", in_signature='', out_signature='as')
@@ -106,6 +99,7 @@ class NotificationFetcher(dbus.service.Object):
     def GetServerInformation(self):
         return ("statnot", "http://code.k2h.se", "0.0.2", "1")
 
+
 if __name__ == '__main__':
     dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
     session_bus = dbus.SessionBus()

osd-clock

 #!/bin/sh
-date '+%Y-%m-%d  %H:%M:%S' | osdc --min-duration=1 $@ -
+LANG=cs_CZ.UTF-8 date '+%a %Y-%m-%d  %H:%M:%S' | osdc --min-duration=1 $@ -

osdd-file.c

@@ -22,6 +22,7 @@ struct file_state
 	char * file_name;
 	char * open_mode;
 	struct osd_to_string_state to_string;
+	bool end_by_null_char;
 };
 
 // *************************  MAIN FUNCTIONS *******
@@ -34,6 +35,8 @@ void file_write(struct file_state *state, VECTOR(char) out)
 	if(f)
 	{
 		fprintf(f,"%s", out);
+		if(state->end_by_null_char)
+			putc(0, f);
 		fclose(f);
 	}
 	VFREE(out);
@@ -49,6 +52,7 @@ void file_new_help(FILE * f)
 	fprintf(f,"\
 Module FILE help:\n\
 -a, --append \t\tAppend to file (do not overwrite)\n\
+-n, --end-by-null-char\tAppend \\0 to end of each message\n\
 " OSDD_TO_STRING_HELP "\
 \n\
 ");
@@ -60,9 +64,10 @@ struct osd_abstract file_new(int argc, char ** argv, Display * nope)
 	(void)nope;
 	argc++;argv--;
 	struct osd_abstract r;
-	static const char short_opts[] = "+b" OSDD_TO_STRING_SHORTOP;
+	static const char short_opts[] = "+b+n" OSDD_TO_STRING_SHORTOP;
 	static const struct option long_opts[] = {
 		{ "append",		no_argument,	NULL,	'b' },
+		{ "end-by-null-char",		no_argument,	NULL,	'n' },
 		OSDD_TO_STRING_LONGOP
 		{ NULL,		0,			NULL,	0   },
 	};
@@ -83,6 +88,9 @@ struct osd_abstract file_new(int argc, char ** argv, Display * nope)
 			case 'b':
 				state->open_mode = "a";
 				break;
+			case 'n':
+				state->end_by_null_char = true;
+				break;
 			default:
 				fprintf(stderr,"Option %c not exist\n\n",opt);
 				file_new_help(stderr);

osdd-run

@@ -5,12 +5,14 @@ killall osdd
 pkill -f not2osd
 ./not2osd &
 tmp=$(mktemp)
+[ -e $XDG_RUNTIME_DIR/osdd_to_i3csstatus ] && rm $XDG_RUNTIME_DIR/osdd_to_i3csstatus
 echo -e "Xft.render: False\nXft.rgba: none" > $tmp
 XENVIRONMENT=$tmp \
 	./osdd -D \
 	DISPLAY_BY_OUTPUTS default nolog display [ -ndisplay%d -Ftimes-48:bold -Ftimes-32:bold ] \
 	FILE default log loglog [ -b -l' ' -m'\n' -T' ' -tiso -cterm ~/.osdd_log  ] \
-	FILE default log loglast [ -l'\n' -tunix -T'\n' -crgbHEX -r ~/.osdd_last ] \
+	FILE default log loglast [ -l'\n' -tunix -T'\n' -crgbHEX  ~/.osdd_last ] \
+	FILE default log loglast [ -l'\n' -tunix -T'\n' -crgbHEX -n $XDG_RUNTIME_DIR/osdd_to_i3csstatus ] \
 	MQTT default mqtt led [ -e'\0' -E -crgbDEC -tunix -r -R -q1 -N \
 		$(cat ~/.secret/mqtt/jug9-all.addres) \
 		$(cat ~/.secret/mqtt/jug9-led-include.user) \

python-osd/osd/__init__.py

+#!/usr/bin/env python3
+"""
+                  On-Screen Display Python Client Library
+             (c) 2022 Jiri Kalvoda <jirikalvoda@kam.mff.cuni.cz>
+
+
+Basic usage:
+============
+
+import osd
+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)

python-osd/pyproject.toml

+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"

python-osd/setup.cfg

+[metadata]
+name = osd
+version = 0.0.1
+author = Jiri Kalvoda
+author_email = jirikalvoda@kam.mff.cuni.cz
+description = On-Screen Display Client
+long_description_content_type = text/plain
+classifiers =
+    Operating System :: OS Independent
+    Programming Language :: Python :: 3
+    Programming Language :: Python :: 3.7
+    Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3.10
+
+[options]
+packages = osd
+python_requires = >=3.7
+zip_safe = False
+install_requires =
+    argparse >= 0.21
+    typing >= 3.0
+    python-xlib
+
+[bdist_wheel]
+universal = 1