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