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
1 change: 1 addition & 0 deletions .python-version
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
3.13
695 changes: 674 additions & 21 deletions LICENSE

Large diffs are not rendered by default.

4 changes: 0 additions & 4 deletions LICENSE_ThirdParty.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,3 @@ https://fedoraproject.org/wiki/Licensing:TCL
# ttkbootstrap
MIT License
https://github.com/israel-dryer/ttkbootstrap/blob/master/LICENSE

# windows-capture-device-list (python-capture-device-list)
MIT License
https://github.com/yushulx/python-capture-device-list/blob/master/LICENSE
18 changes: 18 additions & 0 deletions Taskfile.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
# https://taskfile.dev

version: '3'


tasks:
debug-get-devices:
cmds:
- uv run debug/get_devices.py
desc: "Debug: Get capture devices"
debug-check-capture:
cmds:
- uv run debug/check_capture.py
desc: "Debug: Capture from a device"
debug-check-writer:
cmds:
- uv run debug/check_writer.py
desc: "Debug: Check video writer"
18 changes: 18 additions & 0 deletions debug/check_capture.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
import cv2

if __name__ == "__main__":
capture = cv2.VideoCapture(6)

try:
while True:
ret, frame = capture.read()
if not ret:
break

cv2.imshow("Webcam", frame)

if cv2.waitKey(1) & 0xFF == ord("q"):
break
finally:
capture.release()
cv2.destroyAllWindows()
42 changes: 42 additions & 0 deletions debug/check_writer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
# ruff: noqa: E402

import sys
from pathlib import Path

import cv2

work_dir = Path(__file__).resolve().parent.parent
sys.path.append(str(work_dir / "src"))

from replayer.file_manager import FileManager
from replayer.internal_types import Resolution
from replayer.writer import Writer

fmt = cv2.VideoWriter.fourcc(*"mp4v")

if __name__ == "__main__":
capture = cv2.VideoCapture(5)

writer = Writer(
FileManager(Path(work_dir / "tmp_video"), 600),
fmt=fmt,
frame_rate=60,
frame_size=Resolution(640, 480),
max_buffer_seconds=10,
)
try:
with writer:
while True:
ret, frame = capture.read()
if not ret:
break

writer.write(frame)

cv2.imshow("Webcam", frame)

if cv2.waitKey(1) & 0xFF == ord("q"):
break
finally:
capture.release()
cv2.destroyAllWindows()
7 changes: 7 additions & 0 deletions debug/get_devices.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
from windows_capture_device_list import list_devices

if __name__ == "__main__":
devices = list_devices()
for device in devices:
print(f"{device.id}: {device.name}")
print(f" - Resolutions: {[f'{res.width}x{res.height}' for res in device.resolutions]}")
41 changes: 41 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
[project]
name = "QuickReplay"
version = "2.0.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"opencv-python>=4.12.0.88",
"ttkbootstrap>=1.18.1",
"windows-capture-device-list",
]
license = { file = "LICENSE" }

[dependency-groups]
dev = ["pyinstaller>=6.16.0", "pytest>=9.0.1", "ruff>=0.14.4", "ty>=0.0.1a25"]

[tool.ruff]
line-length = 120
lint.fixable = ["ALL"]
lint.select = ["ALL"]
lint.ignore = [
"A002",
"EM101",
"EM102",
"FIX002",
"S101",
"TRY003",
"TRY300",
"COM812",
"RUF002",
"D",
"TD",
]


[tool.ruff.per-file-ignores]
"test/*" = ["INP001", "PLR2004", "SLF001"]
"debug/*" = ["INP001", "T201"]

[tool.uv.sources]
windows-capture-device-list = { git = "https://github.com/Nanahuse/windows-capture-device-list.git" }
3 changes: 0 additions & 3 deletions requirements.txt

This file was deleted.

49 changes: 0 additions & 49 deletions src/capture_device.py

This file was deleted.

Empty file added src/replayer/__init__.py
Empty file.
56 changes: 56 additions & 0 deletions src/replayer/capture_wrapper.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
import cv2
from cv2.typing import MatLike


class CaptureWrapper:
def __init__(self, capture: cv2.VideoCapture) -> None:
self._capture: cv2.VideoCapture = capture

self._frame_num = int(self._capture.get(cv2.CAP_PROP_FRAME_COUNT))
self._cursor: int = 0

def is_opened(self) -> bool:
return self._capture.isOpened()

def frame_count(self) -> int:
return self._frame_num

def cursor(self) -> int:
return self._cursor

def has_reached_end(self) -> bool:
return self._cursor == self._frame_num

def read(self) -> MatLike | None:
if self.has_reached_end():
return None

self._cursor += 1
_, frame = self._capture.read()
return frame

def move_first(self) -> None:
self._cursor = 0
self._capture.set(cv2.CAP_PROP_POS_FRAMES, 0)

def move_last(self) -> None:
self._cursor = self._frame_num - 1
self._capture.set(cv2.CAP_PROP_POS_FRAMES, self._cursor)

def move_end(self) -> None:
self._cursor = self._frame_num
self._capture.set(cv2.CAP_PROP_POS_FRAMES, 0) # set to 0 to avoid issues

def move_diff(self, diff: int) -> int:
self._cursor += diff
if self._cursor < 0:
remain = self._cursor
self.move_first()
return remain
if self._cursor >= self._frame_num:
remain = self._cursor - self._frame_num
self.move_end()
return remain

self._capture.set(cv2.CAP_PROP_POS_FRAMES, self._cursor)
return 0
53 changes: 53 additions & 0 deletions src/replayer/file_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
from __future__ import annotations

from collections import deque
from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from pathlib import Path

VIDEO_NAME_EXTENSION: str = "mp4"


@dataclass
class RecordFile:
path: Path
frame_num: int


class FileManager:
def __init__(self, work_dir: Path, frame_threshold: int) -> None:
self._work_dir: Path = work_dir
self._frame_threshold: int = frame_threshold

self._file_queue: deque[RecordFile] = deque()
self._frame_count: int = 0
self._index: int = 0

@property
def frame_count(self) -> int:
return self._frame_count

def files(self) -> list[RecordFile]:
return list(self._file_queue)

def get_new_file_path(self) -> Path:
file_path = self._work_dir / f"record_{self._index}.{VIDEO_NAME_EXTENSION}"
self._index += 1
return file_path

def push_file(self, path: Path, frame_num: int) -> None:
self._file_queue.append(RecordFile(path, frame_num))
self._frame_count += frame_num

while self._frame_count - self._file_queue[0].frame_num >= self._frame_threshold:
self._drop_oldest_file()
if not self._file_queue:
break

def _drop_oldest_file(self) -> None:
record_file = self._file_queue.popleft()

record_file.path.unlink(missing_ok=True)
self._frame_count -= record_file.frame_num
37 changes: 37 additions & 0 deletions src/replayer/internal_types.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from windows_capture_device_list import Resolution as WinResolution

FourCC = int


@dataclass
class Resolution:
x: int
y: int

def to_string(self) -> str:
return f"{self.x} x {self.y}"

@staticmethod
def from_string(s: str) -> Resolution:
x_str, y_str = s.split(" x ")
return Resolution(int(x_str), int(y_str))

def to_tuple(self) -> tuple[int, int]:
return (self.x, self.y)

@staticmethod
def from_win_resolution(res: WinResolution) -> Resolution:
return Resolution(res.width, res.height)


@dataclass
class CaptureDevice:
device_num: int
name: str
resolution: list[Resolution]
Loading