Custom notification command + disabling notifications

README.md

@@ -336,6 +336,14 @@ speed = 2
 
 ; Limit replayed terminal inactivity to max n seconds, default: off
 idle_time_limit = 1
+
+[notifications]
+
+; Should desktop notifications be enabled, default: yes
+enabled = no
+
+; Custom notification command
+command = tmux display-message "$TEXT"
 ```
 
 A very minimal config file could look like that:

asciinema/commands/record.py

@@ -5,6 +5,7 @@ import tempfile
 import asciinema.recorder as recorder
 import asciinema.asciicast.raw as raw
 import asciinema.asciicast.v2 as v2
+import asciinema.notifier as notifier
 from asciinema.api import APIError
 from asciinema.commands.command import Command
 
@@ -25,6 +26,7 @@ class RecordCommand(Command):
         self.overwrite = args.overwrite
         self.raw = args.raw
         self.writer = raw.writer if args.raw else v2.writer
+        self.notifier = notifier.get_notifier(config.notifications_enabled, config.notifications_command)
         self.env = env
 
     def execute(self):
@@ -75,7 +77,8 @@ class RecordCommand(Command):
                 command_env=self.env,
                 capture_env=vars,
                 rec_stdin=self.rec_stdin,
-                writer=self.writer
+                writer=self.writer,
+                notifier=self.notifier
             )
         except v2.LoadError:
             self.print_error("can only append to asciicast v2 format recordings")

asciinema/config.py

@@ -118,6 +118,14 @@ class Config:
     def play_speed(self):
         return self.config.getfloat('play', 'speed', fallback=1.0)
 
+    @property
+    def notifications_enabled(self):
+        return self.config.getboolean('notifications', 'enabled', fallback=True)
+
+    @property
+    def notifications_command(self):
+        return self.config.get('notifications', 'command', fallback=None)
+
 
 def get_config_home(env=os.environ):
     env_asciinema_config_home = env.get("ASCIINEMA_CONFIG_HOME")

asciinema/notifier.py

@@ -51,12 +51,29 @@ class TerminalNotifier(Notifier):
             return ['terminal-notifier', '-title', 'asciinema', '-message', text]
 
 
+class CustomCommandNotifier(Notifier):
+    def __init__(self, command):
+        Notifier.__init__(self)
+        self.command = command
+
+    def notify(self, text):
+        args = ['/bin/sh', '-c', self.command]
+        env = os.environ.copy()
+        env['TEXT'] = text
+        env['ICON_PATH'] = self.get_icon_path()
+        subprocess.run(args, env=env, capture_output=True)
+
+
 class NoopNotifier():
     def notify(self, text):
         pass
 
 
-def get_notifier():
+def get_notifier(enabled=True, command=None):
+    if enabled:
+        if command is not None:
+            return CustomCommandNotifier(command)
+        else:
             for c in [TerminalNotifier, AppleScriptNotifier, LibNotifyNotifier]:
                 n = c()
 

asciinema/recorder.py

@@ -4,13 +4,12 @@ import time
 import asciinema.asciicast.v2 as v2
 import asciinema.pty as pty
 import asciinema.term as term
-import asciinema.notifier as notifier
 from asciinema.async_worker import async_worker
 
 
 def record(path, command=None, append=False, idle_time_limit=None,
            rec_stdin=False, title=None, metadata=None, command_env=None,
-           capture_env=None, writer=v2.writer, record=pty.record):
+           capture_env=None, writer=v2.writer, record=pty.record, notifier=None):
     if command is None:
         command = os.environ.get('SHELL') or 'sh'
 
@@ -46,7 +45,7 @@ def record(path, command=None, append=False, idle_time_limit=None,
         time_offset = v2.get_duration(path)
 
     with async_writer(writer, path, full_metadata, append) as w:
-        with async_notifier() as n:
+        with async_notifier(notifier) as n:
             record(
                 ['sh', '-c', command],
                 w,
@@ -83,15 +82,16 @@ class async_writer(async_worker):
 
 
 class async_notifier(async_worker):
-    def __init__(self):
+    def __init__(self, notifier):
         async_worker.__init__(self)
-        self.notifier = notifier.get_notifier()
+        self.notifier = notifier
 
     def notify(self, text):
         self.enqueue(text)
 
     def perform(self, text):
         try:
+            if self.notifier:
                 self.notifier.notify(text)
         except:
             # we catch *ALL* exceptions here because we don't want failed

tests/config_test.py

@@ -187,3 +187,22 @@ def test_play_idle_time_limit():
 
     config = create_config("[play]\nmaxwait = 2.35")
     assert_equal(2.35, config.play_idle_time_limit)
+
+
+def test_notifications_enabled():
+    config = create_config('')
+    assert_equal(True, config.notifications_enabled)
+
+    config = create_config("[notifications]\nenabled = yes")
+    assert_equal(True, config.notifications_enabled)
+
+    config = create_config("[notifications]\nenabled = no")
+    assert_equal(False, config.notifications_enabled)
+
+
+def test_notifications_command():
+    config = create_config('')
+    assert_equal(None, config.notifications_command)
+
+    config = create_config('[notifications]\ncommand = tmux display-message "$TEXT"')
+    assert_equal('tmux display-message "$TEXT"', config.notifications_command)