344 lines
11 KiB
Python
344 lines
11 KiB
Python
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from fabric import Service, Signal
|
|
from fabric.core.service import Property
|
|
from fabric.utils import (
|
|
Gio,
|
|
exec_shell_command,
|
|
exec_shell_command_async,
|
|
logger,
|
|
time,
|
|
)
|
|
|
|
from utils.functions import is_app_running, run_command
|
|
|
|
|
|
class ScreenCapture(Service):
|
|
"""Service for screen capture and recording functionality"""
|
|
|
|
@Signal
|
|
def screenshot_taken(self, path: str) -> None:
|
|
"""Signal emitted when screenshot is taken."""
|
|
|
|
@Signal
|
|
def recording_started(self, path: str) -> None:
|
|
"""Signal emitted when recording starts."""
|
|
|
|
@Signal
|
|
def recording_stopped(self, path: str) -> None:
|
|
"""Signal emitted when recording stops."""
|
|
|
|
_instance = None
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.home = Path.home()
|
|
self.screenshots_dir = self.home / "Pictures" / "Screenshots"
|
|
self.recordings_dir = self.home / "Videos" / "Recordings"
|
|
self.recording_file = Path("/tmp/recording.txt")
|
|
self.recording_start_time_file = Path("/tmp/recording_start_time.txt")
|
|
|
|
self.screenshots_dir.mkdir(parents=True, exist_ok=True)
|
|
self.recordings_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def notify_send(self, title, message, icon=None, actions=None):
|
|
cmd = ["notify-send", "-a", "Modus"]
|
|
|
|
if icon:
|
|
cmd.extend(["-i", str(icon)])
|
|
|
|
if actions:
|
|
for action in actions:
|
|
cmd.extend(["-A", f"{action}={action}"])
|
|
|
|
cmd.extend([title, message])
|
|
return run_command(cmd)
|
|
|
|
def send_screenshot_notification(self, file_path=None):
|
|
cmd = ["notify-send"]
|
|
cmd.extend(
|
|
[
|
|
"-A",
|
|
"files=Show in Files",
|
|
"-A",
|
|
"view=View",
|
|
"-A",
|
|
"edit=Edit",
|
|
"-i",
|
|
"camera-photo-symbolic",
|
|
"-a",
|
|
"Fabric Screenshot Utility",
|
|
"-h",
|
|
f"STRING:image-path:{file_path}",
|
|
"Screenshot Saved",
|
|
f"Saved Screenshot at {file_path}",
|
|
]
|
|
if file_path
|
|
else ["Screenshot Sent to Clipboard"]
|
|
)
|
|
|
|
proc: Gio.Subprocess = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.STDOUT_PIPE)
|
|
|
|
def do_callback(process: Gio.Subprocess, task: Gio.Task):
|
|
try:
|
|
_, stdout, stderr = process.communicate_utf8_finish(task)
|
|
except Exception:
|
|
logger.error(
|
|
f"[SCREENSHOT] Failed read notification action with error {stderr}"
|
|
)
|
|
return
|
|
|
|
match stdout.strip("\n"):
|
|
case "files":
|
|
exec_shell_command_async(f"xdg-open {self.screenshots_dir}")
|
|
case "view":
|
|
exec_shell_command_async(f"xdg-open {file_path}")
|
|
case "edit":
|
|
exec_shell_command_async(f"satty -f {file_path}")
|
|
|
|
proc.communicate_utf8_async(None, None, do_callback)
|
|
self.screenshot_taken(file_path)
|
|
|
|
def send_recording_notification(self, file_path):
|
|
cmd = ["notify-send"]
|
|
cmd.extend(
|
|
[
|
|
"-A",
|
|
"files=Show in Files",
|
|
"-A",
|
|
"view=View",
|
|
"-i",
|
|
"camera-video-symbolic",
|
|
"-a",
|
|
"Fabric Screenshot Utility",
|
|
"Screencast Saved",
|
|
f"Saved Screencast at {file_path}",
|
|
]
|
|
)
|
|
|
|
proc: Gio.Subprocess = Gio.Subprocess.new(cmd, Gio.SubprocessFlags.STDOUT_PIPE)
|
|
|
|
def do_callback(process: Gio.Subprocess, task: Gio.Task):
|
|
try:
|
|
_, stdout, stderr = process.communicate_utf8_finish(task)
|
|
except Exception:
|
|
logger.error(
|
|
f"[SCREENCAST] Failed read notification action with error {stderr}"
|
|
)
|
|
return
|
|
|
|
match stdout.strip("\n"):
|
|
case "files":
|
|
exec_shell_command_async(f"xdg-open {self.recordings_dir}")
|
|
case "view":
|
|
exec_shell_command_async(f"xdg-open {file_path}")
|
|
|
|
proc.communicate_utf8_async(None, None, do_callback)
|
|
self.recording_stopped(file_path)
|
|
|
|
def check_wf_recorder(self):
|
|
if not is_app_running("wf-recorder"):
|
|
return False
|
|
|
|
run_command(["pkill", "-INT", "-x", "wf-recorder"])
|
|
|
|
if self.recording_file.exists():
|
|
recording_file = self.recording_file.read_text().strip()
|
|
self.send_recording_notification(recording_file)
|
|
|
|
if self.recording_start_time_file.exists():
|
|
self.recording_start_time_file.unlink()
|
|
|
|
return True
|
|
|
|
def get_default_monitor_source(self) -> str | None:
|
|
"""Return the monitor source of the default sink (system audio).
|
|
|
|
wf-recorder defaults to the mic input. To capture the audio coming
|
|
out of the speakers (e.g. a Zoom meeting), we need the monitor of
|
|
the default sink (e.g. ``alsa_output.pci-...monitor``).
|
|
"""
|
|
try:
|
|
result = run_command(["pactl", "get-default-sink"])
|
|
if result.returncode == 0:
|
|
sink = (result.stdout or "").strip()
|
|
if sink:
|
|
return f"{sink}.monitor"
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
# PipeWire fallback: parse `wpctl status` for the default sink
|
|
try:
|
|
result = run_command(["wpctl", "status"])
|
|
if result.returncode == 0:
|
|
lines = (result.stdout or "").splitlines()
|
|
in_sinks = False
|
|
default_id = None
|
|
sinks: dict[str, str] = {}
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
if stripped.startswith("Sinks:"):
|
|
in_sinks = True
|
|
continue
|
|
if in_sinks and stripped and not stripped.startswith("─"):
|
|
if "Sources:" in stripped or "Filters:" in stripped:
|
|
in_sinks = False
|
|
continue
|
|
if stripped.startswith("*"):
|
|
default_id = stripped.lstrip("*").strip().split(".")[0]
|
|
parts = stripped.lstrip("* ").split(".", 1)
|
|
if len(parts) == 2:
|
|
sinks[parts[0]] = parts[1].split(" [")[0].strip()
|
|
if default_id and default_id in sinks:
|
|
return f"{sinks[default_id]}.monitor"
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
return None
|
|
|
|
def record_video(self, output_file, *args):
|
|
cmd = [
|
|
"wf-recorder",
|
|
*args,
|
|
"-f",
|
|
str(output_file),
|
|
"-c",
|
|
"libvpx-vp9",
|
|
"--pixel-format",
|
|
"yuv420p",
|
|
"-F",
|
|
"eq=brightness=0.12:contrast=1.1",
|
|
]
|
|
return run_command(cmd)
|
|
|
|
def record_video_noaudio(self, output_file, *args):
|
|
cmd = [
|
|
"wf-recorder",
|
|
*args,
|
|
"-f",
|
|
str(output_file),
|
|
"-c",
|
|
"libvpx-vp9",
|
|
"--pixel-format",
|
|
"yuv420p",
|
|
"-F",
|
|
"eq=brightness=0.12:contrast=1.1",
|
|
"--no-audio",
|
|
]
|
|
return run_command(cmd)
|
|
|
|
def screenshot(self, fullscreen=False, save_copy=True):
|
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
|
|
command = [
|
|
"hyprshot",
|
|
"-s",
|
|
"-o",
|
|
str(self.screenshots_dir),
|
|
"-f",
|
|
f"{timestamp}_fabric_screenshot.png",
|
|
]
|
|
|
|
command.extend(
|
|
["-m", "active", "-m", "output"] if fullscreen else ["-m", "region"]
|
|
)
|
|
command.append("-- ls") # This is so there is something in stdout
|
|
|
|
try:
|
|
exec_shell_command_async(
|
|
" ".join(command),
|
|
self._after_screenshot,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Screenshot failed: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def _after_screenshot(self, *_):
|
|
try:
|
|
screenshot_files = list(self.screenshots_dir.glob("*.png"))
|
|
if screenshot_files:
|
|
latest_file = max(screenshot_files, key=lambda f: f.stat().st_mtime)
|
|
self.send_screenshot_notification(file_path=str(latest_file))
|
|
time.sleep(0.2)
|
|
else:
|
|
# No file found after command: likely user cancelled the selection
|
|
self.notify_send(
|
|
"Screenshot cancelled",
|
|
"Selection was cancelled",
|
|
icon="camera-photo-symbolic",
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Screenshot notification failed: {e}")
|
|
|
|
def record(self, fullscreen=False, no_audio=False):
|
|
if self.is_recording:
|
|
logger.error(
|
|
"[SCREENRECORD] Another instance of wf-recorder is already running"
|
|
)
|
|
return
|
|
|
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
output_file = self.recordings_dir / f"{timestamp}.mp4"
|
|
|
|
self._current_recording_path = str(output_file)
|
|
self.recording_file.write_text(str(output_file))
|
|
self.recording_start_time_file.write_text(str(int(time.time())))
|
|
|
|
if fullscreen:
|
|
area = ""
|
|
else:
|
|
geometry = exec_shell_command("slurp")
|
|
if not geometry or not str(geometry).strip():
|
|
self.notify_send(
|
|
"Recording cancelled",
|
|
"Selection was cancelled",
|
|
icon="camera-video-symbolic",
|
|
)
|
|
return False
|
|
area = f"-g '{geometry}'"
|
|
|
|
if no_audio:
|
|
command = f"wf-recorder --file={output_file} --pixel-format yuv420p --no-audio {area}"
|
|
else:
|
|
monitor = self.get_default_monitor_source()
|
|
if monitor:
|
|
logger.info(f"[SCREENRECORD] Recording system audio from: {monitor}")
|
|
command = (
|
|
f"wf-recorder --file={output_file} --pixel-format yuv420p "
|
|
f"--audio={monitor} {area}"
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"[SCREENRECORD] No system monitor source found, "
|
|
"falling back to default audio source (mic)"
|
|
)
|
|
command = (
|
|
f"wf-recorder --file={output_file} --pixel-format yuv420p "
|
|
f"--audio {area}"
|
|
)
|
|
|
|
exec_shell_command_async(command)
|
|
self.recording_started(str(output_file))
|
|
|
|
return True
|
|
|
|
def stop_recording(self):
|
|
exec_shell_command_async("killall -INT wf-recorder")
|
|
self.recording_stopped(self._current_recording_path)
|
|
return True
|
|
|
|
@Property(bool, "readable", default_value=False)
|
|
def is_recording(self):
|
|
return is_app_running("wf-recorder")
|
|
|
|
|
|
screen_capture_service = ScreenCapture()
|