Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ You can run the app like any other desktop app on your computer. If you decided

### Recording

From the app tray or GUI, you can start and stop a recording as well as pause and resume a recording. Pausing and resuming is important for when you want to hide sensitive information like credit card of login credentials. You can optionally name your recording and give it a description upon stopping a recording. You can also view your recordings by pressing the "Show Recordings" option.
From the app tray or GUI, you can start and stop a recording as well as pause and resume a recording. Pausing and resuming is important for when you want to hide sensitive information like credit card of login credentials. You can optionally name your recording and give it a description upon stopping a recording. You can also view your recordings by pressing the "Show Recordings" option. Alternatively, press `ctrl`+`alt`+`r` from anywhere to start or stop a recording.

### Playback

Expand Down
34 changes: 32 additions & 2 deletions ducktrack/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,15 @@
import sys
from platform import system

from PyQt6.QtCore import QTimer, pyqtSlot
from PyQt6.QtCore import QSettings, QTimer, pyqtSignal, pyqtSlot
from PyQt6.QtGui import QAction, QIcon
from PyQt6.QtWidgets import (QApplication, QCheckBox, QDialog, QFileDialog,
QFormLayout, QLabel, QLineEdit, QMenu,
QMessageBox, QPushButton, QSystemTrayIcon,
QTextEdit, QVBoxLayout, QWidget)

from pynput.keyboard import GlobalHotKeys

from .obs_client import close_obs, is_obs_running, open_obs
from .playback import Player, get_latest_recording
from .recorder import Recorder
Expand Down Expand Up @@ -43,6 +45,10 @@ def get_values(self):
return self.title_input.text(), self.description_input.toPlainText()

class MainInterface(QWidget):
# emitted by the hotkey listener thread so that toggle_record
# always runs on the Qt main thread
toggle_record_requested = pyqtSignal()

def __init__(self, app: QApplication):
super().__init__()
self.tray = QSystemTrayIcon(QIcon(resource_path("assets/duck.png")))
Expand All @@ -52,10 +58,33 @@ def __init__(self, app: QApplication):

self.init_tray()
self.init_window()


if system() == "Darwin":
self.show_macos_permissions_notice()

if not is_obs_running():
self.obs_process = open_obs()

self.toggle_record_requested.connect(self.toggle_record)
self.hotkey_listener = GlobalHotKeys({"<ctrl>+<alt>+r": self.toggle_record_requested.emit})
self.hotkey_listener.start()

def show_macos_permissions_notice(self):
settings = QSettings("TheDuckAI", "DuckTrack")
if settings.value("shown_macos_permissions_notice", False, type=bool):
return
QMessageBox.information(
self,
"Permissions Required",
"DuckTrack needs several macOS permissions to record correctly:\n\n"
"1. Screen Recording (for OBS) - to capture your screen\n"
"2. Accessibility - to track mouse movements\n"
"3. Input Monitoring - to track keyboard events\n\n"
"If recordings come out empty, check System Settings > Privacy & Security "
"and make sure DuckTrack and OBS are allowed."
)
settings.setValue("shown_macos_permissions_notice", True)

def init_window(self):
self.setWindowTitle("DuckTrack")
layout = QVBoxLayout(self)
Expand Down Expand Up @@ -170,6 +199,7 @@ def quit(self):
self.toggle_record()
if hasattr(self, "obs_process"):
close_obs(self.obs_process)
self.hotkey_listener.stop()
self.app.quit()

def closeEvent(self, event):
Expand Down
27 changes: 23 additions & 4 deletions ducktrack/keycomb.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,32 +2,51 @@

from .util import name_to_key

# pynput reports left/right variants (e.g. ctrl_l) on some platforms,
# so fold them into the generic modifier for combination matching
CANONICAL_KEY_NAMES = {
"ctrl_l": "ctrl", "ctrl_r": "ctrl",
"alt_l": "alt", "alt_r": "alt", "alt_gr": "alt",
"shift_l": "shift", "shift_r": "shift",
"cmd_l": "cmd", "cmd_r": "cmd",
}

class KeyCombinationListener:
"""
Simple and bad key combination listener.
"""

def __init__(self):
self.current_keys = set()
self.callbacks = {}
self.listener = Listener(on_press=self.on_key_press, on_release=self.on_key_release)

def add_comb(self, keys, callback):
self.callbacks[tuple([name_to_key(key_name) for key_name in sorted(keys)])] = callback
# store combos in canonical form so they compare against canonicalized presses
self.callbacks[tuple([self.canonicalize_key(name_to_key(key_name)) for key_name in sorted(keys)])] = callback

def canonicalize_key(self, key):
# fold left/right modifier variants first, then let listener.canonical
# undo the effect of held modifiers on the reported character
# (e.g. ctrl+r arriving as '\x12')
name = getattr(key, "name", None)
if name in CANONICAL_KEY_NAMES:
key = name_to_key(CANONICAL_KEY_NAMES[name])
return self.listener.canonical(key)

def on_key_press(self, key):
self.current_keys.add(key)
self.current_keys.add(self.canonicalize_key(key))
for comb, callback in self.callbacks.items():
if all(k in self.current_keys for k in comb):
return callback()

def on_key_release(self, key):
key = self.canonicalize_key(key)
if key in self.current_keys:
self.current_keys.remove(key)

def start(self):
self.listener.start()

def stop(self):
self.listener.stop()
self.listener.stop()
73 changes: 55 additions & 18 deletions ducktrack/obs_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import time
from platform import system

from obsws_python.error import OBSSDKRequestError
import obsws_python as obs
import psutil

Expand All @@ -17,7 +18,19 @@ def is_obs_running() -> bool:
raise Exception("Could not check if OBS is running already. Please check manually.")

def close_obs(obs_process: subprocess.Popen):
if obs_process:
if system() == "Darwin":
# OBS is launched with `open -a`, so obs_process is not the OBS process
# itself and terminating it would do nothing - quit OBS via AppleScript
try:
subprocess.run(["osascript", "-e", 'tell application "OBS" to quit'], timeout=10, check=False)
except subprocess.TimeoutExpired:
pass
for _ in range(10):
if not is_obs_running():
return
time.sleep(0.5)
subprocess.run(["killall", "OBS"], check=False)
elif obs_process:
obs_process.terminate()
try:
obs_process.wait(timeout=5)
Expand Down Expand Up @@ -64,6 +77,10 @@ def open_obs() -> subprocess.Popen:
# you have to change the working directory first for OBS to find the correct locale on windows
os.chdir(os.path.dirname(obs_path))
obs_path = os.path.basename(obs_path)
if system() == "Darwin":
# `open -a` launches OBS the way macOS expects, which keeps the
# screen recording & accessibility permission prompts working
return subprocess.Popen(["open", "-a", "OBS", "--args", "--startreplaybuffer", "--minimize-to-tray"])
return subprocess.Popen([obs_path, "--startreplaybuffer", "--minimize-to-tray"])
except:
raise Exception("Failed to find OBS, please open OBS manually.")
Expand All @@ -83,10 +100,22 @@ def __init__(
output_height=720,
):
self.metadata = metadata

self.req_client = obs.ReqClient()
self.event_client = obs.EventClient()


# OBS may still be starting up (e.g. when DuckTrack just launched it),
# so retry the websocket connection until it is ready to take requests
max_attempts = 5
for attempt in range(1, max_attempts + 1):
try:
self.req_client = obs.ReqClient()
self.event_client = obs.EventClient()
print("connected to OBS version:", self.req_client.get_version().obs_version)
break
except Exception as e:
if attempt == max_attempts:
raise Exception("Could not connect to OBS. Please check that it is running and that the websocket server is enabled.") from e
print(f"waiting for OBS websocket (attempt {attempt}/{max_attempts}): {e}")
time.sleep(2)

self.record_state_events = {}

def on_record_state_changed(data):
Expand All @@ -98,15 +127,19 @@ def on_record_state_changed(data):

self.event_client.callback.register(on_record_state_changed)

self.old_profile = self.req_client.get_profile_list().current_profile_name

if "computer_tracker" not in self.req_client.get_profile_list().profiles:
self.req_client.create_profile("computer_tracker")
else:
self.req_client.set_current_profile("computer_tracker")
self.req_client.create_profile("temp")
self.req_client.remove_profile("temp")
self.req_client.set_current_profile("computer_tracker")
self.old_profile = None
try:
self.old_profile = self.req_client.get_profile_list().current_profile_name

if "computer_tracker" not in self.req_client.get_profile_list().profiles:
self.req_client.create_profile("computer_tracker")
else:
self.req_client.set_current_profile("computer_tracker")
self.req_client.create_profile("temp")
self.req_client.remove_profile("temp")
self.req_client.set_current_profile("computer_tracker")
except OBSSDKRequestError as e:
print(f"warning: could not switch to the computer_tracker profile, continuing with the current profile: {e}")

base_width = metadata["screen_width"]
base_height = metadata["screen_height"]
Expand Down Expand Up @@ -149,16 +182,20 @@ def on_record_state_changed(data):

try:
self.req_client.set_input_mute("Mic/Aux", muted=True)
except obs.error.OBSSDKRequestError :
# In case there is no Mic/Aux input, this will throw an error
except OBSSDKRequestError:
pass
# In case there is no Mic/Aux input, this will throw an error

def start_recording(self):
self.req_client.start_record()

def stop_recording(self):
self.req_client.stop_record()
self.req_client.set_current_profile(self.old_profile) # restore old profile
if self.old_profile:
try:
self.req_client.set_current_profile(self.old_profile) # restore old profile
except OBSSDKRequestError as e:
print(f"warning: could not restore the previous OBS profile: {e}")

def pause_recording(self):
self.req_client.pause_record()
Expand Down Expand Up @@ -197,4 +234,4 @@ def _scale_resolution(base_width: int, base_height: int, target_width: int, tar
scaled_height = int((target_area / aspect_ratio) ** 0.5)
scaled_width = int(aspect_ratio * scaled_height)

return scaled_width, scaled_height
return scaled_width, scaled_height
57 changes: 57 additions & 0 deletions tests/test_keycomb.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
from pynput.keyboard import HotKey, Key, KeyCode

from ducktrack.keycomb import KeyCombinationListener


def make_listener(keys):
listener = KeyCombinationListener()
fired = []
listener.add_comb(keys, lambda: fired.append(1))
return listener, fired


def test_combo_fires_with_generic_modifier():
listener, fired = make_listener(("shift", "esc"))
listener.on_key_press(Key.shift)
listener.on_key_press(Key.esc)
assert fired == [1]


def test_combo_fires_with_left_right_modifier_variants():
listener, fired = make_listener(("shift", "esc"))
listener.on_key_press(Key.shift_l)
listener.on_key_press(Key.esc)
assert fired == [1]


def test_release_clears_pressed_state():
listener, fired = make_listener(("shift", "esc"))
listener.on_key_press(Key.shift_l)
listener.on_key_release(Key.shift_l)
listener.on_key_press(Key.esc)
assert fired == []
assert listener.current_keys == {listener.canonicalize_key(Key.esc)}


def test_shifted_character_is_canonicalized():
# with shift held, pynput reports 'R' instead of 'r'
listener, fired = make_listener(("shift", "r"))
listener.on_key_press(Key.shift)
listener.on_key_press(KeyCode.from_char("R"))
assert fired == [1]


def test_non_matching_combo_does_not_fire():
listener, fired = make_listener(("shift", "esc"))
listener.on_key_press(Key.ctrl)
listener.on_key_press(Key.esc)
assert fired == []


def test_record_hotkey_spec_parses_and_fires():
# same spec as the GlobalHotKeys binding in app.py
fired = []
hotkey = HotKey(HotKey.parse("<ctrl>+<alt>+r"), lambda: fired.append(1))
for key in [Key.ctrl, Key.alt, KeyCode.from_char("r")]:
hotkey.press(key)
assert fired == [1]