Use libtmux instead of own tmux wrapper (#43)

This should make interfacing tmux much more stable and reduce testing requirements.

catmux/catmux_create_session.py

@@ -33,6 +33,8 @@
 
 from importlib import resources
 
+import libtmux
+
 from catmux.session import Session as CatmuxSession
 import catmux.exceptions
 
@@ -82,33 +84,6 @@ def main():
     """Creates a new tmux session if it does not yet exist"""
     args = parse_arguments()
 
-    session_config = CatmuxSession(
-        server_name=args.server_name,
-        session_name=args.session_name,
-        runtime_params=args.overwrite,
-    )
-    try:
-        session_config.init_from_filepath(args.session_config)
-    except catmux.exceptions.InvalidConfig as err:
-        logging.error(err)
-        sys.exit(1)
-
-    try:
-        subprocess.check_call(
-            ["tmux", "-L", args.server_name, "has-session", "-t", args.session_name]
-        )
-        print(
-            'Session with name "{}" already exists. Not overwriting session.'.format(
-                args.session_name
-            )
-        )
-        sys.exit(0)
-    except subprocess.CalledProcessError:
-        # When has-session returns non-zero exit code, the session already exists or there is
-        # probably something severely wrong. TODO: This could be done better probably
-        pass
-
-    command = ["tmux", "-L", args.server_name]
     if args.tmux_config:
         tmux_config = args.tmux_config
         if not os.path.exists(tmux_config):
@@ -123,18 +98,31 @@ def main():
             ".".join(["catmux", "resources"]), "tmux_default.conf"
         ) as catmux_session:
             tmux_config = str(catmux_session)
-
     print("Using tmux config file: {}".format(tmux_config))
-    command += ["-f", tmux_config]
 
-    command += ["new-session", "-s", args.session_name]
-    command.append("-d")
-    print(" ".join(command))
-    if not safe_call(command):
+    tmux_server = libtmux.Server(args.server_name, config_file=tmux_config)
+    try:
+        tmux_server.sessions.get(session_name=args.session_name)
+        print(
+            f'Session with name "{args.session_name}" already exists. Not overwriting session.'
+        )
         sys.exit(1)
+    except libtmux.exc.ObjectDoesNotExist:
+        # If no session with that name can be found, everything is fine...
+        pass
 
-    print('Created session "{}"'.format(args.session_name))
+    session_config = CatmuxSession(
+        session_name=args.session_name,
+        runtime_params=args.overwrite,
+    )
+    try:
+        session_config.init_from_filepath(args.session_config)
+        print(f'Created session "{args.session_name}"')
 
-    session_config.run()
-    if not args.detach:
-        safe_call(["tmux", "-L", args.server_name, "attach", "-t", args.session_name])
+        target_session = session_config.run(tmux_server)
+        if not args.detach:
+            tmux_server.attach_session(target_session=args.session_name)
+    except Exception as err:
+        logging.error(err)
+        tmux_server.kill_session(args.session_name)
+        sys.exit(1)

catmux/session.py

@@ -25,10 +25,10 @@
 
 """Contains everything around the config file"""
 import re
+import libtmux
 import yaml
 
 from catmux.window import Window
-import catmux.tmux_wrapper as tmux
 import catmux.exceptions as cme
 
 
@@ -42,11 +42,10 @@ class Session(object):
 
     """Parser for a config yaml file"""
 
-    def __init__(self, server_name, session_name, runtime_params=None):
+    def __init__(self, session_name: str, runtime_params=None):
         """TODO: to be defined1."""
+        self.session_name = session_name
 
-        self._server_name = server_name
-        self._session_name = session_name
         self._parameters = dict()
         self._runtime_params = self._parse_overwrites(runtime_params)
         self._windows = list()
@@ -75,26 +74,24 @@ def init_from_yaml(self, yaml_data):
         self._parse_parameters()
         self._parse_windows()
 
-    def run(self, debug=False):
+    def run(self, parent_server: libtmux.Server, debug=False):
         """Runs the loaded session"""
         assert len(self._windows) > 0
 
+        tmux_session = parent_server.new_session(self.session_name, attach=False)
+
         first = True
         for window in self._windows:
-            window.create(first)
+            window.run(tmux_session, first)
             if debug:
                 window.debug()
             first = False
 
-        tmux_wrapper = tmux.TmuxWrapper(self._server_name)
         if self._default_window:
-            tmux_wrapper.tmux_call(
-                [
-                    "select-window",
-                    "-t",
-                    self._session_name + ":" + self._default_window,
-                ]
-            )
+            target_window = tmux_session.windows.get(window_name=self._default_window)
+            target_window.select_window()
+
+        return tmux_session
 
     def _parse_common(self):
         if self.__yaml_data is None:
@@ -228,8 +225,6 @@ def _parse_windows(self):
 
                 print(kwargs)
 
-                self._windows.append(
-                    Window(self._server_name, self._session_name, **kwargs)
-                )
+                self._windows.append(Window(**kwargs))
         else:
             raise cme.InvalidConfig("No window section found in session config")

catmux/split.py

@@ -25,7 +25,7 @@
 
 """A split in a tmux session"""
 
-import catmux.tmux_wrapper as tmux
+import libtmux
 
 
 class Split(object):
@@ -46,8 +46,7 @@ def debug(self, name="", prefix=""):
             print(prefix + "  commands: ")
             print("\t- " + "\n\t- ".join(getattr(self, "commands")))
 
-    def run(self, server_name, target_window=None):
+    def run(self, parent_pane: libtmux.Pane):
         "Executes all configured commands" ""
-        tmux_wrapper = tmux.TmuxWrapper(server_name=server_name)
         for command in getattr(self, "commands"):
-            tmux_wrapper.send_keys(command, target_window)
+            parent_pane.send_keys(command)

catmux/window.py

@@ -26,7 +26,8 @@
 """Contains the Window object"""
 import time
 
-import catmux.tmux_wrapper as tmux
+import libtmux
+
 from catmux.split import Split
 from catmux.exceptions import InvalidConfig
 
@@ -35,12 +36,9 @@ class Window(object):
 
     """Class to represent a tmux window structure"""
 
-    def __init__(self, server_name, session_name, **kwargs):
+    def __init__(self, **kwargs):
         """TODO: to be defined1."""
 
-        self.server_name = server_name
-        self.session_name = session_name
-
         split_list = kwargs.pop("splits", None)
         if not split_list:
             if "commands" in kwargs:
@@ -72,28 +70,28 @@ def debug(self):
         for counter, split in enumerate(self.splits):
             split.debug(name=str(counter), prefix=" ")
 
-    def create(self, first=False):
+    def run(self, parent_session: libtmux.Session, first=False):
         """Creates the window"""
-        tmux_wrapper = tmux.TmuxWrapper(server_name=self.server_name)
-        target_window = ":".join([self.session_name, getattr(self, "name")])
+        window_name = getattr(self, "name")
+        target_window = None
         if not first:
-            tmux_wrapper.tmux_call(["new-window", "-t", self.session_name + ":"])
-        tmux_wrapper.tmux_call(
-            ["rename-window", "-t", f"{self.session_name}:$", getattr(self, "name")]
-        )
+            target_window = parent_session.new_window(window_name)
+        else:
+            target_window = parent_session.windows[0]
+            target_window.rename_window(window_name)
+
         for counter, split in enumerate(self.splits):
             if counter > 0:
-                tmux_wrapper.split(target_window)
+                target_window.split_window()
+            target_pane = target_window.panes[-1]
 
             if hasattr(self, "before_commands"):
                 for cmd in getattr(self, "before_commands"):
-                    tmux_wrapper.send_keys(cmd, target_window=target_window)
-            split.run(server_name=self.server_name, target_window=target_window)
+                    target_pane.send_keys(cmd)
 
+            split.run(target_pane)
         if hasattr(self, "layout"):
-            tmux_wrapper.tmux_call(
-                ["select-layout", "-t", target_window, getattr(self, "layout")]
-            )
+            target_window.select_layout(getattr(self, "layout"))
 
         if hasattr(self, "delay"):
             print(

setup.py

@@ -25,7 +25,7 @@
         "License :: OSI Approved :: MIT License",
         "Operating System :: OS Independent",
     ],
-    install_requires=["pyyaml"],
+    install_requires=["pyyaml", "libtmux"],
     python_requires=">=3.7",
     package_data={"catmux.resources": ["*.yaml", "*.txt", "*.conf"]},
 )

test/test_helpers.py

@@ -1,7 +1,6 @@
 import os
 
 from catmux.session import check_boolean_field
-from catmux.tmux_wrapper import _safe_call as safe_call
 from catmux.prefix import get_prefix
 import catmux.resources
 
@@ -26,12 +25,6 @@ def test_boolean_field():
     assert check_boolean_field(False) == False
 
 
-def test_safe_call():
-    assert safe_call(["bash", "-c", "exit 0"]) == True
-    assert safe_call(["bash", "-c", "exit 1"]) == False
-    assert safe_call(["bash", "-c", "return 0"]) == False  #  illegal bash
-
-
 def test_get_prefix():
     prefix = os.path.join(get_prefix(), "__init__.py")
     assert prefix == catmux.resources.__file__