Add watchdog functionality

qutil/git.py

@@ -5,16 +5,28 @@ Created on Mon Aug 14 18:36:43 2023
 @author: Simon Humpohl
 """
 import autologging
-from typing import Set, Dict, Optional
+from typing import Set, Dict, Optional, List
 from dataclasses import dataclass
 from pathlib import Path
 import tkinter.messagebox
 from subprocess import check_output, check_call
 
+
 try:
-    import git
+    import watchdog.events
 except ImportError:
-    git = None
+    watchdog = None
+    EventHandler = None
+else:
+    class EventHandler(watchdog.events.FileSystemEventHandler):
+        def __init__(self, watch_dog: 'GitWatchDog'):
+            # TODO: Come up with an even more infantile name
+            self._doggo = watch_dog
+
+        def on_any_event(self, event):
+            if event.event_type == watchdog.events.EVENT_TYPE_CLOSED:
+                return
+            self._doggo._possible_change = True
 
 
 @autologging.logged
@@ -40,7 +52,8 @@ class SimpleRepo:
     def push(self):
         check_call(['git', 'push'], cwd=self.repo_path)
 
-    def get_submodules(self):
+    def get_submodules(self) -> List['SimpleRepo']:
+        """Get a list of all submodules this repository has."""
         sms = check_output(['git', 'submodule'], cwd=self.repo_path).decode('utf8').splitlines()
         folders = []
         for sm in sms:
@@ -49,27 +62,28 @@ class SimpleRepo:
         return folders
 
     def get_current_hash(self) -> bytes:
+        """Get the hash of the repository HEAD via "git rev-parse HEAD"."""
         hash_base64, = check_output(['git', 'rev-parse', 'HEAD'], cwd=self.repo_path).splitlines()
         return hash_base64
 
-
-def auto_commit(repo: SimpleRepo):
+    def auto_commit(self):
+        """Stage all changes and commit them with a default commit message."""
         msg = 'Automatic commit. Lets hope this is nothing that needs explanation.'
-    repo.add_all()
-    repo.commit_staged_changes(msg)    
-
+        self.add_all()
+        self.commit_staged_changes(msg)
 
-def manual_commit(repo: SimpleRepo):
-    while status := repo.get_status():
-        title = f'Manual commit: {repo.name}'
+    def request_manual_commit(self):
+        """Bother the user with an dialogue box to commit all changes"""
+        while status := self.get_status():
+            title = f'Manual commit: {self.name}'
 
-        msg = (f'The repository {repo.name} is dirty. '
+            msg = (f'The repository {self.name} is dirty. '
                    'Commit, ignore or stash the changes and press OK. '
                    'Pressing cancel will raise a KeyboardInterrupt.\n\n'
                    f'{status}')
 
             if not tkinter.messagebox.askokcancel(title, msg):
-            raise KeyboardInterrupt(repo.name)
+                raise KeyboardInterrupt(self.name)
 
 
 @dataclass
@@ -79,19 +93,68 @@ class GitRules:
     auto_push: Dict[Path, Optional]
 
     def process(self, repo: SimpleRepo):
+        if repo.repo_path in self.ignore:
+            return
+
         if repo.get_status():
             for sm in repo.get_submodules():
                 self.process(sm)
 
             if repo.get_status():
-                if repo.repo_path in auto_commit:
-                    auto_commit(repo)
+                if repo.repo_path in self.auto_commit:
+                    repo.auto_commit()
                 else:
-                    manual_commit(repo)
+                    repo.request_manual_commit()
 
         if repo.repo_path in self.auto_push:
             push_rule = self.auto_push[repo.repo_path]
             if push_rule is None:
                 repo.push()
             else:
-                raise NotImplementedError("TODO")
+                raise NotImplementedError("TODO: implement rules which branch to push to which repository here")
+
+
+class GitWatchDog:
+    def __init__(self, repo: SimpleRepo, rules: GitRules, watch: bool, on_ipython_cell_exec: bool):
+        self.rules = rules
+        self.repo = repo
+        self._observer = None
+        self._possible_change = None
+        self._ipython_register = None
+        self._event_handler = None
+        self.watch = watch
+        self.on_ipython_cell_exec = on_ipython_cell_exec
+
+    @property
+    def watch(self) -> bool:
+        return self._observer is not None
+
+    @watch.setter
+    def watch(self, watch: bool):
+        if watch and not self.watch:
+            import watchdog.observers
+            if self._event_handler is None:
+                self._event_handler = EventHandler(self)
+            self._observer = watchdog.observers.Observer()
+            self._observer.add_handler_for_watch(self._event_handler, self.repo.repo_path, recursive=True)
+        elif not watch and self.watch:
+            self._observer.stop()
+            self._observer = None
+
+    @property
+    def on_ipython_cell_exec(self) -> bool:
+        return self._ipython_register
+
+    def _check(self):
+        if not self.watch or self._possible_change:
+            self.rules.process(self.repo)
+
+    @on_ipython_cell_exec.setter
+    def on_ipython_cell_exec(self, val: bool):
+        import IPython.core.events
+        ip = IPython.get_ipython()
+        if val and not self.on_ipython_cell_exec:
+            ip.events.register('pre_run_cell', self._check)
+        elif not val and self.on_ipython_cell_exec:
+            ip.events.unregister('pre_run_cell', self._check)
+        self._ipython_register = val