feat(cli): add cli options for changing config

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.9"
 
 dependencies = [
   "obsws-python",
+  "tomli-w",
   "watchdog",
   "tqdm"
 ]

requirements.txt

@@ -1,4 +1,5 @@
 obsws-python==1.8.0
+tomli-w==1.1.0
 tqdm==4.67.2
 watchdog==6.0.0
 websocket-client==1.9.0

rewind/cli.py

@@ -4,6 +4,7 @@ import argparse
 
 from rewind.core import clip, mark, save, print_markers
 from rewind.autostart import install
+from rewind.config import Config
 
 def build_clip_parser(subparsers: argparse._SubParsersAction) -> None:
     clip_parser = subparsers.add_parser("clip", help="Clips the last 'x' seconds")
@@ -49,6 +50,17 @@ def build_install_parser(subparsers: argparse._SubParsersAction) -> None:
         description="Install and enable the rewind daemon using system autostart."
     )
 
+def build_config_parser(subparsers: argparse._SubParsersAction) -> None:
+    config_parser = subparsers.add_parser("config", help="View or edit config values")
+    config_sub = config_parser.add_subparsers(dest="config_command")
+
+    get_parser = config_sub.add_parser("get", help="Print a single config value")
+    get_parser.add_argument("key", help="Dotted key, e.g. obs.host")
+
+    set_parser = config_sub.add_parser("set", help="Update a config value")
+    set_parser.add_argument("key", help="Dotted key, e.g. obs.host")
+    set_parser.add_argument("value", help="New value")
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="rewind",
@@ -62,12 +74,14 @@ def build_parser() -> argparse.ArgumentParser:
     build_list_parser(sub)
     build_save_parser(sub)
     build_install_parser(sub)
+    build_config_parser(sub)
 
     return parser
 
 def main(argv=None) -> int:
     parser = build_parser()
     args = parser.parse_args(argv)
+    cfg = Config()
 
     try:
         if args.command == "clip":
@@ -80,6 +94,13 @@ def main(argv=None) -> int:
             save(args.start, args.end)
         elif args.command == "install":
             install()
+        elif args.command == "config":
+            if args.config_command == "set":
+                cfg.set(args.key, args.value)
+            elif args.config_command == "get":
+                print(cfg.get(args.key))
+            else:
+                cfg.show()
         else:
             parser.error("Unknown command")
     except Exception as e:

rewind/config.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+import os
+import tomllib
+import tomli_w
+from pathlib import Path
+
+from rewind.paths import get_config_dir
+
+_CONFIG_NAME = "config.toml"
+_EXAMPLE_CONFIG = Path(__file__).parent / "example.config.toml"
+
+
+class Config:
+    def __init__(self):
+        self._config_dir = get_config_dir()
+        self._config_file = self._config_dir / _CONFIG_NAME
+        self._data = self._load()
+
+    def _load(self) -> dict:
+        if not self._config_file.exists():
+            self._config_dir.mkdir(parents=True, exist_ok=True)
+            self._config_file.write_text(_EXAMPLE_CONFIG.read_text())
+
+        with self._config_file.open("rb") as f:
+            return tomllib.load(f)
+
+    def _save(self) -> None:
+        with self._config_file.open("wb") as f:
+            tomli_w.dump(self._data, f)
+
+    @property
+    def obs_host(self) -> str:
+        return self._data["obs"]["host"]
+
+    @obs_host.setter
+    def obs_host(self, value: str) -> None:
+        self._data["obs"]["host"] = value
+        self._save()
+
+    @property
+    def obs_port(self) -> int:
+        return self._data["obs"]["port"]
+
+    @obs_port.setter
+    def obs_port(self, value: int) -> None:
+        self._data["obs"]["port"] = value
+        self._save()
+
+    @property
+    def obs_password(self) -> str:
+        return self._data["obs"]["password"]
+
+    @obs_password.setter
+    def obs_password(self, value: str) -> None:
+        self._data["obs"]["password"] = value
+        self._save()
+
+    @property
+    def max_record_time(self) -> int:
+        return self._data["record"]["max_record_time"]
+
+    @max_record_time.setter
+    def max_record_time(self, value: int) -> None:
+        self._data["record"]["max_record_time"] = value
+        self._save()
+
+    @property
+    def clip_output(self) -> str:
+        return os.path.expanduser(self._data["record"]["clip_output"])
+
+    @clip_output.setter
+    def clip_output(self, value: str) -> None:
+        self._data["record"]["clip_output"] = value
+        self._save()
+
+    @property
+    def vod_output(self) -> str:
+        return os.path.expanduser(self._data["record"]["vod_output"])
+
+    @vod_output.setter
+    def vod_output(self, value: str) -> None:
+        self._data["record"]["vod_output"] = value
+        self._save()
+
+    def get(self, key: str) -> str:
+        section, _, field = key.partition(".")
+        try:
+            return str(self._data[section][field])
+        except KeyError:
+            raise KeyError(f"Unknown config key: {key!r}")
+
+    def set(self, key: str, value: str) -> None:
+        section, _, field = key.partition(".")
+        try:
+            current = self._data[section][field]
+        except KeyError:
+            raise KeyError(f"Unknown config key: {key!r}")
+        self._data[section][field] = type(current)(value)
+        self._save()
+
+    def show(self) -> None:
+        for section, values in self._data.items():
+            for field, value in values.items():
+                print(f"{section}.{field} = {value!r}")

rewind/core.py

@@ -4,14 +4,14 @@ import datetime
 import subprocess
 
 from rewind.state import load_state, add_marker_to_state, remove_marker_from_state
-from rewind.paths import load_config
+from rewind.config import Config
 from tqdm import tqdm
 
 def clip(seconds_from_end: float) -> None:
     if seconds_from_end <= 0 or seconds_from_end > 600:
         raise ValueError("Clip length must be positive and less than or equal to 10 minutes")
 
-    clip_output = os.path.expanduser(load_config()["record"]["clip_output"])
+    clip_output = Config().clip_output
     os.makedirs(clip_output, exist_ok=True)
 
     output_file_name = f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}.mp4"
@@ -30,7 +30,7 @@ def clip(seconds_from_end: float) -> None:
     print(f"Created clip: {output_path}")
 
 def save(first_marker: str, second_marker: str):
-    vod_dir = os.path.expanduser(load_config()["record"]["vod_output"])
+    vod_dir = Config().vod_output
     os.makedirs(vod_dir, exist_ok=True)
 
     first_timestamp = get_marker_timestamp(first_marker)

rewind/daemon.py

@@ -3,14 +3,13 @@ import os
 import datetime
 import time
 import obsws_python as obs
-import subprocess
 import logging
-import shutil
 import signal
 
 from watchdog.observers import Observer
 from watchdog.events import FileSystemEventHandler
-from rewind.paths import load_config, get_state_dir
+from rewind.paths import get_state_dir
+from rewind.config import Config
 from rewind.core import mark, marker_exists, remove_marker
 from rewind.state import add_file_to_state, create_state_file_if_needed, cleanup_state
 
@@ -18,7 +17,7 @@ running = True
 LOG_FILE =  get_state_dir() / "daemon.log"
 INTERVAL = 10
 SENTINEL_FILE = os.path.expanduser("~/.config/obs-studio/.sentinel")
-OBS_MAX_RETRIES = 10
+OBS_MAX_RETRIES = 5
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@@ -33,25 +32,6 @@ def shutdown(signum, frame):
 signal.signal(signal.SIGINT, shutdown)
 signal.signal(signal.SIGTERM, shutdown)
 
-def clean_obs_sentinel_files():
-    if os.path.exists(SENTINEL_FILE):
-        try:
-            shutil.rmtree(SENTINEL_FILE)
-            logger.info("Removed existing OBS .sentinel directory")
-        except Exception as e:
-            logger.error(f"Could not delete OBS .sentinel directory: {e}")
-
-def is_obs_running():
-    result = subprocess.run(
-        ["pgrep", "-x", "obs"],
-        stdout=subprocess.PIPE
-    )
-    return result.returncode == 0
-
-def open_obs():
-    # Using and not checking OBS since it needs to be non-blocking
-    subprocess.Popen(["obs", "--minimize-to-tray"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-
 def open_obs_connection(host: str, port: int, password: str, obs_max_retries: int) -> obs.ReqClient:
     con = None
     init_sleep = 1
@@ -107,15 +87,8 @@ class Handler(FileSystemEventHandler):
         add_file_to_state(event.src_path)
 
 def main() -> None:
-    if is_obs_running():
+    config = Config()
-        logger.info("OBS is already running")
+    con = open_obs_connection(config.obs_host, config.obs_port, config.obs_password, OBS_MAX_RETRIES)
-    else:
-        logger.info("OBS is not running, starting it now")
-        clean_obs_sentinel_files()
-        open_obs()
-        
-    config = load_config()
-    con = open_obs_connection(config["obs"]["host"], config["obs"]["port"], config["obs"]["password"], OBS_MAX_RETRIES)
 
     recording_dir = con.get_record_directory().record_directory
     start_recording(con)
@@ -130,8 +103,8 @@ def main() -> None:
         observer.start()
 
         while running:
-            cleanup_physical_files(recording_dir, config["record"]["max_record_time"])
+            cleanup_physical_files(recording_dir, config.max_record_time)
-            cleanup_state(config["record"]["max_record_time"])
+            cleanup_state(config.max_record_time)
             time.sleep(INTERVAL)
     finally:
         if observer:

rewind/paths.py

@@ -1,32 +1,15 @@
 #!/usr/bin/env python3
 import os
-import tomllib
 from pathlib import Path
-from importlib import resources
 
 APP_NAME = "rewind"
-CONFIG_NAME = "config.toml"
 
 def get_config_dir() -> Path:
     base = os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")
     return Path(base) / APP_NAME
 
-def load_config() -> dict:
-    config_dir = get_config_dir()
-    config_file = config_dir / CONFIG_NAME
-
-    # Create config directory if it doesn't exist and replace with default config
-    if not config_file.exists():
-        config_dir.mkdir(parents=True, exist_ok=True)
-        with resources.open_text("rewind", CONFIG_NAME) as default_config:
-            with config_file.open("w") as f:
-                f.write(default_config.read())
-
-    with config_file.open("rb") as f:
-        return tomllib.load(f)
-    
 def get_state_dir() -> Path:
     base = os.path.expanduser("~/.local/share")
     path = Path(base) / APP_NAME
     path.mkdir(parents=True, exist_ok=True)
-    return path
+    return path