git.scottworley.com Git - srec/blob - srec.py

   1 # srec: A simple GUI for screen recording
   2 #
   3 # This program is free software: you can redistribute it and/or modify it
   4 # under the terms of the GNU General Public License as published by the
   5 # Free Software Foundation, version 3.
   6
   7 from __future__ import annotations
   8
   9 from dataclasses import dataclass
  10 from datetime import datetime
  11 import os
  12 import subprocess
  13 import sys
  14 import time
  15 from typing import Any, Callable
  16
  17 import gi
  18 gi.require_version("Gtk", "4.0")
  19 gi.require_version("GLib", "2.0")
  20
  21 from gi.repository import Gtk   # nopep8 pylint: disable=wrong-import-position
  22 from gi.repository import GLib  # nopep8 pylint: disable=wrong-import-position
  23
  24
  25 @dataclass
  26 class Stream:
  27     process: subprocess.Popen[bytes]
  28
  29     @staticmethod
  30     def start(command: list[str]) -> Stream:
  31         # pylint: disable=consider-using-with
  32         return Stream(process=subprocess.Popen(command, stdin=subprocess.PIPE))
  33
  34     def stop(self) -> None:
  35         stdin = self.process.stdin
  36         assert stdin is not None
  37         try:
  38             stdin.write(b'q')
  39             stdin.flush()
  40         except BrokenPipeError:
  41             print("Stream already stopped?", file=sys.stderr)
  42         self.process.wait()
  43
  44
  45 recording: Stream | None = None
  46 sharing: Stream | None = None
  47
  48
  49 def make_filename() -> str:
  50     directory = os.environ.get(
  51         'XDG_VIDEOS_DIR',
  52         os.path.expanduser('~/Videos/SRec'))
  53     os.makedirs(directory, exist_ok=True)
  54     timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  55     return os.path.join(directory, f'srec {timestamp}.mkv')
  56
  57
  58 def video_source(stack: Gtk.Stack) -> list[str]:
  59     if stack.get_child_by_name('not_recording').get_first_child().get_active():
  60         return ['-f', 'x11grab', '-i', ':0.0+0,0']
  61     if is_sharing_enabled(stack):
  62         return ['-f', 'v4l2', '-i', '/dev/video9']
  63     return ['-f', 'v4l2', '-i', '/dev/video0']
  64
  65
  66 def is_sharing_enabled(stack: Gtk.Stack) -> bool:
  67     enabled = stack.get_child_by_name('not_recording').get_first_child(
  68     ).get_next_sibling().get_next_sibling().get_active()
  69     assert isinstance(enabled, bool)
  70     return enabled
  71
  72
  73 def find_size_display(stack: Gtk.Stack) -> Gtk.Label:
  74     return stack.get_child_by_name(
  75         'recording').get_first_child().get_next_sibling()
  76
  77
  78 def summarize_size(n: int) -> str:
  79     if n > 100_000_000:
  80         m = int(n / (1024 * 1024))
  81         return f'{m}M'
  82     if n > 100_000:
  83         k = int(n / 1024)
  84         return f'{k}K'
  85     return str(n)
  86
  87
  88 def begin_monitoring_file_size(size_display: Gtk.Label, filename: str) -> None:
  89     def update_size_display() -> Any:
  90         done = recording is None
  91         if done:
  92             size_display.set_label('')
  93         else:
  94             try:
  95                 size = summarize_size(os.stat(filename).st_size)
  96             except FileNotFoundError:
  97                 size = '--'
  98             size_display.set_label(f'<big>{size}</big>')
  99         return GLib.SOURCE_REMOVE if done else GLib.SOURCE_CONTINUE
 100     GLib.timeout_add_seconds(1, update_size_display)
 101
 102
 103 def on_start_button(_: Gtk.Button, stack: Gtk.Stack) -> None:
 104     global recording, sharing  # pylint: disable=global-statement
 105     assert recording is None
 106     assert sharing is None
 107
 108     filename = make_filename()
 109     begin_monitoring_file_size(find_size_display(stack), filename)
 110
 111     stack.set_visible_child_name("recording")
 112
 113     if is_sharing_enabled(stack):
 114         out_opts = [
 115             '-f', 'v4l2', '-framerate', '25', '-codec:v', 'rawvideo',
 116             '-pix_fmt', 'yuv420p',
 117         ]
 118         sharing = Stream.start(
 119             ['ffmpeg', '-i', '/dev/video0']
 120             + out_opts + ['/dev/video8']
 121             + out_opts + ['/dev/video9'])
 122         time.sleep(3)  # Bad!!  We should not be sleeping in this thread!
 123
 124     recording = Stream.start(
 125         ['ffmpeg', '-framerate', '25'] + video_source(stack) +
 126         ['-f', 'pulse', '-ac', '2', '-i', 'default', filename])
 127
 128
 129 def stop_streams() -> None:
 130     global recording, sharing  # pylint: disable=global-statement
 131     assert recording is not None
 132     recording.stop()
 133     recording = None
 134     if sharing is not None:
 135         sharing.stop()
 136         sharing = None
 137
 138
 139 def on_stop_button(_: Gtk.Button, stack: Gtk.Stack) -> None:
 140     stop_streams()
 141     stack.set_visible_child_name("not_recording")
 142
 143
 144 def make_button(label: str, action: Callable[[
 145                 Gtk.Button, Gtk.Stack], None], stack: Gtk.Stack) -> Gtk.Button:
 146     button = Gtk.Button(label=label)
 147     button.connect('clicked', action, stack)
 148     button.set_margin_top(10)
 149     button.set_margin_start(10)
 150     button.set_margin_end(10)
 151     button.set_margin_bottom(10)
 152     return button
 153
 154
 155 def make_share_control() -> Gtk.CheckButton:
 156     can_share = os.path.exists('/sys/module/v4l2loopback')
 157     control = Gtk.CheckButton(
 158         label='Share Webcam', sensitive=can_share, active=can_share)
 159     control.set_margin_start(20)
 160     return control
 161
 162
 163 def on_activate(app: Gtk.Application) -> None:
 164     win = Gtk.ApplicationWindow(application=app)
 165     win.set_title('SRec')
 166     win.set_icon_name('srec')
 167
 168     stack = Gtk.Stack()
 169
 170     nr_box = Gtk.Box()
 171     nr_box.set_orientation(Gtk.Orientation.VERTICAL)
 172     screen = Gtk.CheckButton(label='Screen')
 173     nr_box.append(screen)
 174     nr_box.append(Gtk.CheckButton(label='Webcam', active=True, group=screen))
 175     nr_box.append(make_share_control())
 176     nr_box.append(make_button("Start Recording", on_start_button, stack))
 177     stack.add_named(nr_box, "not_recording")
 178
 179     r_box = Gtk.Box()
 180     r_box.set_orientation(Gtk.Orientation.VERTICAL)
 181     r_box.append(make_button("Stop Recording", on_stop_button, stack))
 182     r_box.append(Gtk.Label(use_markup=True, justify=Gtk.Justification.CENTER))
 183     stack.add_named(r_box, "recording")
 184
 185     win.set_child(stack)
 186     win.present()
 187
 188
 189 def main() -> None:
 190     app = Gtk.Application(application_id='net.chkno.srec')
 191     app.connect('activate', on_activate)
 192     app.run(None)
 193
 194
 195 if __name__ == '__main__':
 196     main()