Initial MVP application skeleton
Add PySide6 camera UI, YOLO/Tesseract detection pipeline, capture metadata, configuration, and project gitignore.
This commit is contained in:
101
app/media.py
Normal file
101
app/media.py
Normal file
@@ -0,0 +1,101 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
|
||||
def timestamp_name() -> str:
|
||||
return datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
|
||||
|
||||
def write_metadata(media_path: Path, metadata: dict[str, Any]) -> Path:
|
||||
json_path = media_path.with_suffix(".json")
|
||||
with json_path.open("w", encoding="utf-8") as metadata_file:
|
||||
json.dump(metadata, metadata_file, indent=2, ensure_ascii=False)
|
||||
metadata_file.write("\n")
|
||||
return json_path
|
||||
|
||||
|
||||
class MediaStore:
|
||||
def __init__(self, config: dict[str, Any], app_config: Any) -> None:
|
||||
self.config = config
|
||||
self.app_config = app_config
|
||||
|
||||
def photo_path(self) -> Path:
|
||||
capture_cfg = self.config["capture"]
|
||||
directory = self.app_config.resolve_path(capture_cfg["photos_dir"])
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
extension = capture_cfg.get("image_extension", "jpg").lstrip(".")
|
||||
return directory / f"{timestamp_name()}.{extension}"
|
||||
|
||||
def video_path(self) -> Path:
|
||||
capture_cfg = self.config["capture"]
|
||||
directory = self.app_config.resolve_path(capture_cfg["videos_dir"])
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
extension = capture_cfg.get("video_extension", "mp4").lstrip(".")
|
||||
return directory / f"{timestamp_name()}.{extension}"
|
||||
|
||||
def save_photo(self, frame_bgr: np.ndarray, metadata: dict[str, Any]) -> Path:
|
||||
path = self.photo_path()
|
||||
cv2.imwrite(str(path), frame_bgr)
|
||||
write_metadata(path, metadata)
|
||||
return path
|
||||
|
||||
|
||||
class VideoRecorder:
|
||||
def __init__(self, config: dict[str, Any], app_config: Any) -> None:
|
||||
self.config = config
|
||||
self.app_config = app_config
|
||||
self.path: Path | None = None
|
||||
self.writer: cv2.VideoWriter | None = None
|
||||
self.started_at: str | None = None
|
||||
|
||||
@property
|
||||
def is_recording(self) -> bool:
|
||||
return self.writer is not None
|
||||
|
||||
def start(self, frame_bgr: np.ndarray) -> Path:
|
||||
if self.writer is not None:
|
||||
raise RuntimeError("Nagrywanie juz trwa")
|
||||
|
||||
capture_cfg = self.config["capture"]
|
||||
self.path = MediaStore(self.config, self.app_config).video_path()
|
||||
h, w = frame_bgr.shape[:2]
|
||||
fps = float(self.config["camera"].get("fps", 30))
|
||||
codec = str(capture_cfg.get("video_codec", "mp4v"))
|
||||
fourcc = cv2.VideoWriter_fourcc(*codec[:4])
|
||||
self.writer = cv2.VideoWriter(str(self.path), fourcc, fps, (w, h))
|
||||
if not self.writer.isOpened():
|
||||
self.writer = None
|
||||
raise RuntimeError("Nie mozna uruchomic zapisu wideo")
|
||||
self.started_at = datetime.now().isoformat(timespec="seconds")
|
||||
self.write(frame_bgr)
|
||||
return self.path
|
||||
|
||||
def write(self, frame_bgr: np.ndarray) -> None:
|
||||
if self.writer is not None:
|
||||
self.writer.write(frame_bgr)
|
||||
|
||||
def stop(self, metadata: dict[str, Any]) -> Path | None:
|
||||
if self.writer is None:
|
||||
return None
|
||||
self.writer.release()
|
||||
self.writer = None
|
||||
path = self.path
|
||||
if path is not None:
|
||||
metadata = {
|
||||
**metadata,
|
||||
"recording": {
|
||||
"started_at": self.started_at,
|
||||
"stopped_at": datetime.now().isoformat(timespec="seconds"),
|
||||
},
|
||||
}
|
||||
write_metadata(path, metadata)
|
||||
self.path = None
|
||||
self.started_at = None
|
||||
return path
|
||||
Reference in New Issue
Block a user