refactor(daemon): use obs-events for file fetching

Instead of depending on watchdog, use the OBS event which is faster.

pyproject.toml

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

requirements.txt

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

rewind/core.py

@@ -1,12 +1,10 @@
 #!/usr/bin/env python3
 import os
 import datetime
-import subprocess
-import tempfile
 
-from rewind.state import load_state, add_marker_to_state, remove_marker_from_state
 from rewind.config import Config
-from tqdm import tqdm
+from rewind.media import get_ts_files, concat_ts_files
+from rewind.state import load_state, add_marker_to_state, remove_marker_from_state
 
 config = Config()
 
@@ -24,12 +22,14 @@ def clip(seconds_from_end: float) -> None:
     end_timestamp = datetime.datetime.now().timestamp()
     length = end_timestamp - start_timestamp
     
-    files, start_offset, end_offset = _get_ts_files(
+    ts_files = [f for f in load_state()["files"] if os.path.exists(f["path"])]
+    files, start_offset, end_offset = get_ts_files(
+        ts_files,
         start_timestamp,
         end_timestamp
     )
 
-    _concat_ts_files(files, start_offset, end_offset, length, output_path)
+    concat_ts_files(files, start_offset, end_offset, length, output_path)
     print(f"Created clip: {output_path}")
 
 def save(first_marker: str, second_marker: str):
@@ -45,12 +45,15 @@ def save(first_marker: str, second_marker: str):
     output_file_name = f"{datetime.datetime.fromtimestamp(first_timestamp).strftime('%Y-%m-%d_%H:%M:%S')}-[{first_marker}-{second_marker}].mp4"
     output_path = os.path.join(vod_dir, output_file_name)
 
-    files, start_offset, end_offset = _get_ts_files(
+    ts_files = [f for f in load_state()["files"] if os.path.exists(f["path"])]
+
+    files, start_offset, end_offset = get_ts_files(
+        ts_files,
         first_timestamp,
         second_timestamp
     )
 
-    _concat_ts_files(files, start_offset, end_offset, second_timestamp - first_timestamp, output_path)
+    concat_ts_files(files, start_offset, end_offset, second_timestamp - first_timestamp, output_path)
     print(f"Created video file: {output_path}")
 
 def mark(name: str) -> None:
@@ -90,107 +93,4 @@ def marker_exists(name: str) -> bool:
         if marker["name"] == name:
             return True
     
-    return False
-
-"""
-Retrieves .ts files recorded between the specified timestamps.
-Returns a list of file paths and extra start and end offsets if needed.
-get_duration() is used as little as possible since it is slow.
-end_timestamp of a file is the start time of the next file.
-"""
-def _get_ts_files(start_timestamp: float, end_timestamp: float) -> tuple[list[str], float, float]:
-    ts_files = [f for f in load_state()["files"] if os.path.exists(f["path"])]
-    selected_files = []
-    start_offset = 0.0
-    end_offset = 0.0
-
-    for i, file_info in enumerate(ts_files):
-        file_start = file_info["timestamp"]
-        file_end = ts_files[i + 1]["timestamp"] if i + 1 < len(ts_files) else get_duration(file_info["path"]) + file_start
-
-        if file_end <= start_timestamp:
-            continue
-        if file_start >= end_timestamp:
-            break
-
-        selected_files.append(file_info["path"])
-
-        if file_start <= start_timestamp < file_end:
-            start_offset = start_timestamp - file_start
-        if file_start < end_timestamp <= file_end:
-            end_offset = file_end - end_timestamp
-
-    return selected_files, start_offset, end_offset
-
-def _concat_ts_files(file_list: list[str], start_offset: float, end_offset: float, length: float, output_file: str) -> None:
-    tmp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False)
-    try:
-        for file_path in file_list:
-            tmp_file.write(f"file '{file_path}'\n")
-        tmp_file.close()
-
-        cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostats", "-progress", "pipe:1"]
-        if start_offset > 0:
-            cmd += ["-ss", str(start_offset)]
-        if end_offset > 0:
-            cmd += ["-t", str(length)]
-        cmd += ["-f", "concat", "-safe", "0", "-i", tmp_file.name, "-c", "copy"]
-        cmd.append(output_file)
-
-        process = subprocess.Popen(
-            cmd,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            text=True,
-            bufsize=1,
-        )
-
-        init_ms_val = 0
-        init_ms_val_set = False
-
-        with tqdm(
-            total=length,
-            unit="s",
-            unit_scale=True,
-            unit_divisor=60,
-            desc="Processing",
-            leave=True,
-        ) as pbar:
-            for line in process.stdout:  # type: ignore[union-attr]
-                line = line.strip()
-
-                if line.startswith("out_time_ms="):
-                    out_time_ms = int(line.split("=")[1])
-
-                    if not init_ms_val_set:
-                        init_ms_val = out_time_ms
-                        init_ms_val_set = True
-                    out_time_ms -= init_ms_val
-
-                    seconds = abs(out_time_ms / 1_000_000)
-                    pbar.n = min(seconds, length)
-                    pbar.refresh()
-
-                elif line == "progress=end":
-                    break
-
-        ret = process.wait()
-        if ret != 0:
-            raise RuntimeError("ffmpeg failed")
-    finally:
-        os.unlink(tmp_file.name)
-
-def get_duration(file_path: str) -> float:
-    result = subprocess.run(
-        ["ffprobe", "-v", "error", "-show_entries",
-         "format=duration", "-of",
-         "default=noprint_wrappers=1:nokey=1", file_path],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        text=True,
-    )
-
-    if result.returncode != 0:
-        raise RuntimeError(f"ffprobe failed for file {file_path}")
-
-    return float(result.stdout.strip())
+    return False

rewind/daemon.py

@@ -6,8 +6,6 @@ import obsws_python as obs
 import logging
 import signal
 
-from watchdog.observers import Observer
-from watchdog.events import FileSystemEventHandler
 from rewind.paths import get_state_dir
 from rewind.config import Config
 from rewind.core import mark, marker_exists, remove_marker
@@ -31,21 +29,26 @@ def shutdown(signum, frame):
 signal.signal(signal.SIGINT, shutdown)
 signal.signal(signal.SIGTERM, shutdown)
 
-def open_obs_connection(host: str, port: int, password: str, obs_max_retries: int) -> obs.ReqClient:
-    con = None
+def open_obs_connection(host: str, port: int, password: str, obs_max_retries: int) -> dict[str, obs.ReqClient | obs.EventClient]:
+    req = None
+    event = None
     init_sleep = 1
 
     for _ in range(obs_max_retries):
         try:
-            con = obs.ReqClient(host=host, port=port, password=password)
+            req = obs.ReqClient(host=host, port=port, password=password)
+            event = obs.EventClient(host=host, port=port, password=password)
         except ConnectionRefusedError:
             logger.info("OBS WebSocket not ready, retrying...")
         except obs.events.OBSSDKError:
             raise RuntimeError("Check OBS credentials")
             
-        if con:
+        if req and event:
             logger.info(f"Successfully connected to OBS at {host}:{port}")
-            return con
+            return {
+                "req": req,
+                "event": event
+            }
 
         time.sleep(init_sleep)
         init_sleep *= 2
@@ -76,42 +79,37 @@ def cleanup_physical_files(directory: str, max_age_seconds: int) -> None:
                 os.remove(file_path)
                 logger.info(f"Removed old file: {file_path}")
 
-class Handler(FileSystemEventHandler):
-    def on_created(self, event):
-        if event.is_directory:
-            return
-        if not event.src_path.endswith(".ts"):
-            return
-
-        add_file_to_state(event.src_path)
+def on_record_file_changed(event) -> None:
+    new_file = event.new_output_path
+    add_file_to_state(new_file)
 
 def main() -> None:
     config = Config()
-    con = open_obs_connection(config.get_obs_host(), config.get_obs_port(), config.get_obs_password(), OBS_MAX_RETRIES)
+    obs_host = config.get_obs_host()
+    obs_port = config.get_obs_port()
+    obs_password = config.get_obs_password()
 
-    recording_dir = con.get_record_directory().record_directory
-    start_recording(con)
+    connections = open_obs_connection(obs_host, obs_port, obs_password, OBS_MAX_RETRIES)
+    req_con = connections["req"]
+    event_con = connections["event"]
+
+    event_con.callback.register(on_record_file_changed)
+
+    recording_dir = req_con.get_record_directory().record_directory
+    start_recording(req_con)
 
     create_state_file_if_needed()
     create_initial_marker()
 
     try:
-        event_handler = Handler()
-        observer = Observer()
-        observer.schedule(event_handler, path=recording_dir, recursive=False)
-        observer.start()
-
         while running:
             cleanup_physical_files(recording_dir, config.get_max_record_time())
             cleanup_state(config.get_max_record_time())
             time.sleep(INTERVAL)
     finally:
-        if observer:
-            observer.stop()
-            observer.join()
-
-        stop_recording(con)
-        con.disconnect()
+        stop_recording(req_con)
+        event_con.disconnect()
+        req_con.disconnect()
         logger.info("Daemon stopped")
 
 if __name__ == "__main__":

rewind/media.py

@@ -0,0 +1,114 @@
+import os
+import subprocess
+import tempfile
+
+from rewind.state import load_state
+from tqdm import tqdm
+
+"""
+Retrieves .ts files recorded between the specified timestamps.
+Returns a list of file paths and extra start and end offsets if needed.
+get_duration() is used as little as possible since it is slow.
+end_timestamp of a file is the start time of the next file.
+"""
+def get_ts_files(
+        ts_files: list[dict],
+        start_timestamp: float, 
+        end_timestamp: float
+        ) -> tuple[list[str], float, float]:
+    selected_files = []
+    start_offset = 0.0
+    end_offset = 0.0
+
+    for i, file_info in enumerate(ts_files):
+        file_start = file_info["timestamp"]
+        file_end = ts_files[i + 1]["timestamp"] if i + 1 < len(ts_files) else get_duration(file_info["path"]) + file_start
+        
+        if file_end <= start_timestamp:
+            continue
+        if file_start >= end_timestamp:
+            break
+
+        selected_files.append(file_info["path"])
+
+        if file_start < start_timestamp:
+            start_offset = start_timestamp - file_start
+
+        if file_end >= end_timestamp:
+            end_offset = file_end - end_timestamp
+            break
+        
+    return selected_files, start_offset, end_offset
+
+def concat_ts_files(file_list: list[str], start_offset: float, end_offset: float, length: float, output_file: str) -> None:
+    tmp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False)
+    try:
+        for file_path in file_list:
+            tmp_file.write(f"file '{file_path}'\n")
+        tmp_file.close()
+
+        cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-nostats", "-progress", "pipe:1"]
+        if start_offset > 0:
+            cmd += ["-ss", str(start_offset)]
+        if end_offset > 0:
+            cmd += ["-t", str(length)]
+        cmd += ["-f", "concat", "-safe", "0", "-i", tmp_file.name, "-c", "copy"]
+        cmd.append(output_file)
+
+        process = subprocess.Popen(
+            cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1,
+        )
+
+        init_ms_val = 0
+        init_ms_val_set = False
+
+        with tqdm(
+            total=length,
+            unit="s",
+            unit_scale=True,
+            unit_divisor=60,
+            desc="Processing",
+            leave=True,
+        ) as pbar:
+            for line in process.stdout:  # type: ignore[union-attr]
+                line = line.strip()
+
+                if line.startswith("out_time_ms="):
+                    out_time_ms = int(line.split("=")[1])
+
+                    if not init_ms_val_set:
+                        init_ms_val = out_time_ms
+                        init_ms_val_set = True
+                    out_time_ms -= init_ms_val
+
+                    seconds = abs(out_time_ms / 1_000_000)
+                    pbar.n = min(seconds, length)
+                    pbar.refresh()
+
+                elif line == "progress=end":
+                    break
+
+        ret = process.wait()
+        if ret != 0:
+            raise RuntimeError("ffmpeg failed")
+    finally:
+        os.unlink(tmp_file.name)
+
+def get_duration(file_path: str) -> float:
+    result = subprocess.run(
+        ["ffprobe", "-v", "error", "-show_entries",
+         "format=duration", "-of",
+         "default=noprint_wrappers=1:nokey=1", file_path],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+
+    if result.returncode != 0:
+        raise RuntimeError(f"ffprobe failed for file {file_path}")
+
+    return float(result.stdout.strip())

rewind/state.py

@@ -37,7 +37,7 @@ def write_state(state: dict) -> None:
 
 def add_file_to_state(file_path: str) -> None:
     state = load_state()
-    files = state.get("files", [])
+    files = sorted(state.get("files", []), key=lambda f: f["timestamp"])
 
     # Idempotency guard
     existing_paths = {f["path"] for f in files}
@@ -46,7 +46,7 @@ def add_file_to_state(file_path: str) -> None:
 
     files.append({
         "path": file_path,
-        "timestamp": datetime.datetime.now().timestamp(),
+        "timestamp": datetime.datetime.now().timestamp()
     })
 
     state["files"] = files