diff --git a/aeonview.py b/aeonview.py index 7d6e664..4d2af2f 100644 --- a/aeonview.py +++ b/aeonview.py @@ -1,3 +1,9 @@ +"""aeonview - Timelapse generator using ffmpeg. + +Downloads webcam images on a schedule and generates daily, monthly, and yearly +timelapse videos by stitching frames together with ffmpeg. +""" + from __future__ import annotations import argparse @@ -5,15 +11,16 @@ import hashlib import logging import subprocess import sys +import tempfile from datetime import datetime, timedelta from pathlib import Path import requests -# Define messages for logging -# These messages are used for logging purposes and can be customized as needed. class AeonViewMessages: + """Constant log messages used throughout the application.""" + INVALID_URL = "Invalid URL provided." INVALID_DATE = "Invalid date format provided." DOWNLOAD_SUCCESS = "Image downloaded successfully." @@ -25,34 +32,37 @@ class AeonViewMessages: class AeonViewImages: - """ - Class to handle image download and saving. + """Handle image download and saving. - This class is responsible for downloading images from a URL and saving them - to a specified directory. + Downloads images from a webcam URL and saves them into a date-based + directory structure under the project path. """ def __init__(self, project_path: Path, url: str | None, args=None): + """Initialize the AeonViewImages class. + + :param project_path: Path to the project directory. + :param url: URL of the image to download. + :param args: Command-line arguments passed to the class. """ - Initialize the AeonViewImages class. - :param project_path: Path to the project directory - :param url: URL of the image to download - :param args: Command line arguments passed to the class - """ - self.project_path = project_path or None + self.project_path = project_path self.url = url or None self.args = args or {} self.simulate = getattr(args, "simulate", False) def get_image_paths( - self, url: str | None, destination_base: Path | None, date: datetime + self, + url: str | None, + destination_base: Path | None, + date: datetime | None, ) -> dict: - """ - Get image paths for saving the downloaded image. - :param url: URL of the image - :param destination_base: Base path where the image will be saved - :param date: Date for which the image is requested - :return: Image object + """Compute image paths for saving the downloaded image. + + :param url: URL of the image. + :param destination_base: Base path where the image will be saved. + :param date: Date for which the image is requested. + :return: Dictionary with url, file, date, and destinations info. + :raises SystemExit: If any parameter is invalid. """ if url is None or not isinstance(url, str): logging.error(AeonViewMessages.INVALID_URL) @@ -87,12 +97,11 @@ class AeonViewImages: file_name = date.strftime("%H-%M-%S") + extension destination_file = AeonViewHelpers.build_path(destination, file_name) - if not destination.exists(): - if getattr(self.args, "simulate", False): - logging.info("Simulate: would create %s", destination) - else: - AeonViewHelpers.mkdir_p(destination) - logging.info("Creating destination base path: %s", destination) + if self.simulate: + logging.info("Simulate: would create %s", destination) + else: + AeonViewHelpers.mkdir_p(destination) + logging.info("Creating destination base path: %s", destination) return { "url": url, @@ -114,8 +123,9 @@ class AeonViewImages: } def get_current_image(self): - """ - Download the image from the URL and save it to the project directory. + """Download the image from the URL and save it to the project directory. + + :raises SystemExit: If the URL is missing or the date format is invalid. """ date_param = getattr(self.args, "date", None) @@ -154,11 +164,12 @@ class AeonViewImages: "Simulate: would download %s to %s", self.url, dest_file ) - def download_image(self, destination: Path): - """ - Download the image using Python's requests library. - :param destination: Path where the image will be saved - :return: None + def download_image(self, destination: Path | str): + """Download the image using Python's requests library. + + :param destination: Path where the image will be saved. + :raises SystemExit: If the URL is missing, destination is not a Path, + or the HTTP request fails. """ if self.url is None: @@ -169,7 +180,7 @@ class AeonViewImages: logging.error("Invalid destination path.") sys.exit(1) - if not getattr(self.args, "simulate", False): + if not self.simulate: logging.info("Downloading image from %s", self.url) response = requests.get(self.url, stream=True, timeout=10) if response.status_code == 200: @@ -191,20 +202,19 @@ class AeonViewImages: class AeonViewVideos: - """ - Class to handle video generation and management. + """Handle video generation and management. - This class is responsible for generating daily, monthly, and yearly videos. - It uses ffmpeg for video processing. + Generates daily timelapse videos from images, then concatenates daily + videos into monthly and yearly compilations using ffmpeg. """ def __init__( self, project_path: Path, args: argparse.Namespace | None = None ): - """ - Initialize the AeonViewVideos class. - :param project_path: Path to the project directory - :param args: Command line arguments passed to the class + """Initialize the AeonViewVideos class. + + :param project_path: Path to the project directory. + :param args: Command-line arguments passed to the class. """ self.project_path = project_path self.args = args @@ -219,8 +229,9 @@ class AeonViewVideos: self.path_videos = AeonViewHelpers.build_path(self.project_path, "vid") def generate_daily_video(self): - """ - Generate a daily video from images. + """Generate a daily timelapse video from images. + + :raises SystemExit: If the input image directory does not exist. """ year_month = f"{self.year}-{self.month}" @@ -242,8 +253,7 @@ class AeonViewVideos: if not input_dir.exists(): logging.error("Input directory %s does not exist", input_dir) sys.exit(1) - if not output_dir.exists(): - AeonViewHelpers.mkdir_p(output_dir) + AeonViewHelpers.mkdir_p(output_dir) subprocess.run(ffmpeg_cmd, check=True) logging.info( "%s: %s", AeonViewMessages.VIDEO_GENERATION_SUCCESS, output_file @@ -251,38 +261,105 @@ class AeonViewVideos: else: logging.info("Simulate: would run %s", " ".join(ffmpeg_cmd)) - def generate_monthly_video(self, output_dir: Path): + def generate_monthly_video(self): + """Generate a monthly video by concatenating daily videos. + + Discovers daily ``.mp4`` files in ``vid/YYYY-MM/`` and concatenates + them, excluding the monthly output file itself to avoid corruption + on re-runs. + + :raises SystemExit: If the input directory is missing or contains + no daily videos. """ - Generate a monthly video from images. - :param output_dir: Directory where the video will be saved - :return: None - """ - raise NotImplementedError( - "Monthly video generation is not implemented." + year_month = f"{self.year}-{self.month}" + output_dir = AeonViewHelpers.build_path(self.path_videos, year_month) + output_file = AeonViewHelpers.build_path( + output_dir, f"{year_month}.mp4" ) - def generate_yearly_video(self, output_dir: Path): + if not output_dir.exists(): + logging.error("Input directory %s does not exist", output_dir) + sys.exit(1) + + daily_videos = sorted( + f for f in output_dir.glob("*.mp4") if f != output_file + ) + if not daily_videos: + logging.error("No daily videos found in %s", output_dir) + sys.exit(1) + + self._concatenate_videos( + f"monthly video for {year_month}", daily_videos, output_file + ) + + def generate_yearly_video(self): + """Generate a yearly video by concatenating monthly videos. + + Discovers monthly ``.mp4`` files across ``vid/YYYY-*/`` directories + and concatenates them into a single yearly video. + + :raises SystemExit: If no monthly videos are found for the year. """ - Generate a yearly video from images. - :param output_dir: Directory where the video will be saved - :return: None + year = self.year + output_dir = AeonViewHelpers.build_path(self.path_videos, year) + output_file = AeonViewHelpers.build_path(output_dir, f"{year}.mp4") + + monthly_videos = sorted(self.path_videos.glob(f"{year}-*/{year}-*.mp4")) + + if not monthly_videos: + logging.error("No monthly videos found for year %s", year) + sys.exit(1) + + self._concatenate_videos( + f"yearly video for {year}", monthly_videos, output_file + ) + + def _concatenate_videos( + self, label: str, input_videos: list[Path], output_file: Path + ) -> None: + """Concatenate multiple video files into one using ffmpeg concat demuxer. + + :param label: Human-readable label for log messages (e.g. "monthly video for 2025-04"). + :param input_videos: List of input video file paths to concatenate. + :param output_file: Path for the resulting concatenated video. """ - raise NotImplementedError("Yearly video generation is not implemented.") + logging.info("Generating %s", label) + logging.info("Output file will be %s", output_file) + + if not self.simulate: + AeonViewHelpers.mkdir_p(output_file.parent) + cmd, concat_file = AeonViewHelpers.generate_concat_command( + input_videos, output_file + ) + logging.info("Running ffmpeg command: %s", " ".join(cmd)) + try: + subprocess.run(cmd, check=True) + logging.info( + "%s: %s", + AeonViewMessages.VIDEO_GENERATION_SUCCESS, + output_file, + ) + finally: + concat_file.unlink(missing_ok=True) + else: + logging.info( + "Simulate: would concatenate %d videos into %s", + len(input_videos), + output_file, + ) class AeonViewHelpers: - """ - Helper class for common operations. - """ + """Utility methods for path manipulation, argument parsing, and ffmpeg commands.""" @staticmethod def check_date(year: int, month: int, day: int) -> bool: - """ - Check if the given year, month, and day form a valid date. - :param year: Year to check - :param month: Month to check - :param day: Day to check - :return: True if valid date, False otherwise + """Check if the given year, month, and day form a valid date. + + :param year: Year to check. + :param month: Month to check. + :param day: Day to check. + :return: True if valid date, False otherwise. """ try: date = datetime(year, month, day) @@ -292,26 +369,28 @@ class AeonViewHelpers: @staticmethod def mkdir_p(path: Path): - """ - Create a directory and all parent directories if they do not exist. - :param path: Path to the directory to create - :return: None + """Create a directory and all parent directories if they do not exist. + + :param path: Path to the directory to create. """ path.mkdir(parents=True, exist_ok=True) @staticmethod def build_path(base: Path, *args) -> Path: - """ - Build a path from the base and additional arguments. - :param base: Base path - :param args: Parts of the path to join - :return: Structured and resolved path + """Build a path from the base and additional arguments. + + :param base: Base path. + :param args: Parts of the path to join. + :return: Structured and resolved path. """ return Path(base).joinpath(*args or []).resolve() @staticmethod def parse_arguments(): - """Parse command line arguments.""" + """Parse command-line arguments. + + :return: Tuple of (parsed Namespace, ArgumentParser instance). + """ dest_default = str(Path.cwd() / "projects") @@ -357,9 +436,10 @@ class AeonViewHelpers: @staticmethod def get_extension(url: str | None) -> str | None: - """ - Get the file extension from the URL. - :return: File extension + """Get the file extension from the URL. + + :param url: URL to extract the extension from. + :return: File extension string, or None if url is None. """ if url is None: logging.error(AeonViewMessages.INVALID_IMAGE_EXTENSION) @@ -379,10 +459,9 @@ class AeonViewHelpers: @staticmethod def setup_logger(verbose: bool): - """ - Set up the logger for the application. - :param verbose: Enable verbose logging if True - :return: None + """Set up the logger for the application. + + :param verbose: Enable verbose logging if True. """ level = logging.DEBUG if verbose else logging.INFO logging.basicConfig(level=level, format="[%(levelname)s] %(message)s") @@ -391,12 +470,12 @@ class AeonViewHelpers: def generate_ffmpeg_command( input_dir: Path, output_file: Path, fps: int = 10 ) -> list: - """ - Generate the ffmpeg command to create a video from images. - :param input_dir: Directory containing the images - :param output_file: Path to the output video file - :param fps: Frames per second for the video - :return: Newly created ffmpeg command as a list + """Generate the ffmpeg command to create a video from images. + + :param input_dir: Directory containing the images. + :param output_file: Path to the output video file. + :param fps: Frames per second for the video. + :return: ffmpeg command as a list of strings. """ return [ "ffmpeg", @@ -414,10 +493,44 @@ class AeonViewHelpers: str(output_file), ] + @staticmethod + def generate_concat_command( + input_files: list[Path], output_file: Path + ) -> tuple[list[str], Path]: + """Generate an ffmpeg concat demuxer command for joining video files. + + :param input_files: List of input video file paths. + :param output_file: Path to the output video file. + :return: Tuple of (ffmpeg command list, temporary concat list file path). + """ + concat_list = tempfile.NamedTemporaryFile( + mode="w", suffix=".txt", delete=False + ) + for f in input_files: + concat_list.write(f"file '{f}'\n") + concat_list.close() + + cmd = [ + "ffmpeg", + "-y", + "-f", + "concat", + "-safe", + "0", + "-i", + concat_list.name, + "-c", + "copy", + str(output_file), + ] + return cmd, Path(concat_list.name) + class AeonViewApp: - """ - Main application class for AeonView. + """Main application class for AeonView. + + Parses arguments and dispatches to image download or video generation + workflows based on the selected mode. """ def __init__(self): @@ -434,9 +547,7 @@ class AeonViewApp: self.base_path = Path(self.args.dest).resolve() def run(self): - """ - Execute the application based on the provided arguments. - """ + """Execute the application based on the provided arguments.""" if self.args.simulate: logging.info("Simulation mode active. No actions will be executed.") @@ -446,8 +557,9 @@ class AeonViewApp: self.process_video() def process_image(self): - """ - Process image download and saving based on the provided arguments. + """Process image download and saving based on the provided arguments. + + :raises SystemExit: If the URL is missing or invalid. """ if not self.args.url or self.args.url is None: logging.error("--url is required in image mode") @@ -471,8 +583,10 @@ class AeonViewApp: avi.get_current_image() def process_video(self): - """ - Process video generation based on the provided arguments. + """Process video generation based on the provided arguments. + + :raises SystemExit: If the project is missing, invalid, or the + generate date cannot be parsed. """ if not self.args.project: logging.error("--project is required in video mode") @@ -511,10 +625,6 @@ class AeonViewApp: self.args.month = month self.args.year = year - if not AeonViewHelpers.check_date(int(year), int(month), int(day)): - logging.error("Invalid date: %s-%s-%s", year, month, day) - sys.exit(1) - args: argparse.Namespace = self.args avm = AeonViewVideos(project_path, args) @@ -522,16 +632,12 @@ class AeonViewApp: if self.args.timeframe == "daily": avm.generate_daily_video() elif self.args.timeframe == "monthly": - output_dir = AeonViewHelpers.build_path( - project_path, "vid", f"{year}-{month}" - ) - avm.generate_monthly_video(output_dir) + avm.generate_monthly_video() elif self.args.timeframe == "yearly": - output_dir = AeonViewHelpers.build_path(project_path, "vid", year) - avm.generate_yearly_video(output_dir) + avm.generate_yearly_video() -if __name__ == "__main__": +if __name__ == "__main__": # pragma: no cover app = AeonViewApp() app.run() diff --git a/aeonview_test.py b/aeonview_test.py index b334c78..60998d3 100644 --- a/aeonview_test.py +++ b/aeonview_test.py @@ -1,5 +1,8 @@ +"""Tests for the aeonview timelapse generator.""" + # vim: set ft=python ts=4 sw=4 sts=4 et wrap: import argparse +import contextlib import logging import subprocess import tempfile @@ -10,6 +13,7 @@ from unittest import mock import pytest from aeonview import ( + AeonViewApp, AeonViewHelpers, AeonViewImages, AeonViewMessages, @@ -27,6 +31,41 @@ default_test_path = Path(tempfile.gettempdir(), "test_project").resolve() tmp_images = Path(tempfile.gettempdir(), "images") +# --------------------------------------------------------------------------- +# Test helpers +# --------------------------------------------------------------------------- + + +def make_video_args( + simulate=False, fps=10, day="01", month="04", year="2025" +) -> argparse.Namespace: + """Create an ``argparse.Namespace`` with video-generation defaults.""" + return argparse.Namespace( + simulate=simulate, fps=fps, day=day, month=month, year=year + ) + + +@contextlib.contextmanager +def expect_error_exit(): + """Context manager that expects ``SystemExit`` and silences ``logging.error``.""" + with pytest.raises(SystemExit), mock.patch("aeonview.logging.error"): + yield + + +def make_app_with_project(tmp: str) -> tuple[AeonViewApp, Path]: + """Create an ``AeonViewApp`` whose base_path points at *tmp* with a 'proj' dir.""" + app = AeonViewApp() + app.base_path = Path(tmp).resolve() + proj_path = app.base_path / "proj" + proj_path.mkdir() + return app, proj_path + + +# --------------------------------------------------------------------------- +# AeonViewHelpers tests +# --------------------------------------------------------------------------- + + def test_build_path_resolves_correctly(): base = Path(tempfile.gettempdir()) result = AeonViewHelpers.build_path(base, "a", "b", "c") @@ -62,6 +101,12 @@ def test_get_extension_invalid(): assert AeonViewHelpers.get_extension(None) is None +def test_get_extension_jpeg(): + assert ( + AeonViewHelpers.get_extension(f"{default_image_domain}.jpeg") == ".jpeg" + ) + + def test_generate_ffmpeg_command(): input_dir = tmp_images output_file = Path(tempfile.gettempdir(), "output.mp4") @@ -94,6 +139,37 @@ def test_simulate_ffmpeg_call(mock_run): mock_run.assert_called_once_with(cmd, check=True) +def test_generate_concat_command(): + with tempfile.TemporaryDirectory() as tmp: + files = [Path(tmp) / "01.mp4", Path(tmp) / "02.mp4"] + for f in files: + f.touch() + output = Path(tmp) / "output.mp4" + cmd, concat_file = AeonViewHelpers.generate_concat_command( + files, output + ) + try: + assert cmd[0] == "ffmpeg" + assert "-f" in cmd + assert "concat" in cmd + assert "-safe" in cmd + assert "0" in cmd + assert "-c" in cmd + assert "copy" in cmd + assert str(output) == cmd[-1] + assert concat_file.exists() + content = concat_file.read_text(encoding="utf-8") + for f in files: + assert f"file '{f}'" in content + finally: + concat_file.unlink(missing_ok=True) + + +# --------------------------------------------------------------------------- +# AeonViewImages tests +# --------------------------------------------------------------------------- + + def test_get_image_paths_valid(): url = f"{default_image_domain}.jpg" destination_base = default_test_path @@ -109,7 +185,7 @@ def test_get_image_paths_valid(): def test_get_image_paths_invalid_url(): - with pytest.raises(SystemExit), mock.patch("aeonview.logging.error"): + with expect_error_exit(): aeon_view_images = AeonViewImages(default_test_path, "invalid-url") aeon_view_images.get_image_paths( "invalid-url", default_test_path, datetime(2025, 4, 10) @@ -117,7 +193,7 @@ def test_get_image_paths_invalid_url(): def test_get_image_paths_invalid_date(): - with pytest.raises(SystemExit), mock.patch("aeonview.logging.error"): + with expect_error_exit(): aeon_view_images = AeonViewImages( default_test_path, f"{default_image_domain}.jpg" ) @@ -129,6 +205,62 @@ def test_get_image_paths_invalid_date(): ) +def test_get_image_paths_none_url(): + avi = AeonViewImages(default_test_path, None) + with expect_error_exit(): + avi.get_image_paths(None, default_test_path, datetime(2025, 4, 10)) + + +def test_get_image_paths_no_image_extension(): + url = "https://example.com/image.bmp" + avi = AeonViewImages(default_test_path, url) + with expect_error_exit(): + avi.get_image_paths(url, default_test_path, datetime(2025, 4, 10)) + + +def test_get_image_paths_none_destination(): + url = f"{default_image_domain}.jpg" + avi = AeonViewImages(default_test_path, url) + with expect_error_exit(): + avi.get_image_paths(url, None, datetime(2025, 4, 10)) + + +def test_get_image_paths_non_path_destination(): + url = f"{default_image_domain}.jpg" + avi = AeonViewImages(default_test_path, url) + with expect_error_exit(): + # noinspection PyTypeChecker + avi.get_image_paths( + url, + "/tmp/not-a-path-object", # pyright: ignore [reportArgumentType] + datetime(2025, 4, 10), + ) + + +def test_get_image_paths_creates_directory(): + with tempfile.TemporaryDirectory() as tmp: + url = f"{default_image_domain}.jpg" + dest = Path(tmp) / "newproject" + avi = AeonViewImages(dest, url) + paths = avi.get_image_paths(url, dest, datetime(2025, 4, 10, 12, 0, 0)) + assert paths["file"] == "12-00-00.jpg" + assert (dest / "2025-04" / "10").exists() + + +def test_get_image_paths_simulate_no_mkdir(): + with tempfile.TemporaryDirectory() as tmp: + url = f"{default_image_domain}.jpg" + dest = Path(tmp) / "simproject" + args = argparse.Namespace(simulate=True) + avi = AeonViewImages(dest, url, args) + with mock.patch("aeonview.logging.info"): + paths = avi.get_image_paths( + url, dest, datetime(2025, 4, 10, 12, 0, 0) + ) + assert paths["file"] == "12-00-00.jpg" + assert not (dest / "2025-04" / "10").exists() + + @mock.patch("aeonview.AeonViewHelpers.mkdir_p") @mock.patch("aeonview.AeonViewImages.download_image") def test_get_current_image(mock_download_image, mock_mkdir_p): @@ -139,6 +271,30 @@ def test_get_current_image(mock_download_image, mock_mkdir_p): mock_download_image.assert_called() +def test_get_current_image_invalid_date_format(): + args = argparse.Namespace(simulate=False, date="not-a-date") + avi = AeonViewImages(default_test_path, f"{default_image_domain}.jpg", args) + with expect_error_exit(): + avi.get_current_image() + + +@mock.patch("aeonview.AeonViewHelpers.mkdir_p") +@mock.patch("aeonview.AeonViewImages.download_image") +def test_get_current_image_no_date_uses_now(mock_download, mock_mkdir): + args = argparse.Namespace(simulate=False, date=None) + avi = AeonViewImages(default_test_path, f"{default_image_domain}.jpg", args) + avi.get_current_image() + mock_mkdir.assert_called() + mock_download.assert_called() + + +def test_get_current_image_no_url(): + args = argparse.Namespace(simulate=False, date="2025-04-10 12:30:45") + avi = AeonViewImages(default_test_path, None, args) + with expect_error_exit(): + avi.get_current_image() + + @mock.patch("aeonview.requests.get") def test_download_image_success(mock_get): mock_response = mock.Mock() @@ -167,17 +323,61 @@ def test_download_image_failure(mock_get): avi = AeonViewImages(default_test_path, f"{default_image_domain}.jpg", args) destination = Path(tempfile.gettempdir(), "image.jpg") - with pytest.raises(SystemExit), mock.patch("aeonview.logging.error"): + with expect_error_exit(): avi.download_image(destination) +def test_download_image_no_url(): + args = argparse.Namespace(simulate=False) + avi = AeonViewImages(default_test_path, None, args) + with expect_error_exit(): + avi.download_image(Path(tempfile.gettempdir(), "image.jpg")) + + +def test_download_image_non_path_destination(): + args = argparse.Namespace(simulate=False) + avi = AeonViewImages(default_test_path, f"{default_image_domain}.jpg", args) + with expect_error_exit(): + # noinspection PyTypeChecker + avi.download_image( + "/tmp/image.jpg" # pyright: ignore [reportArgumentType] + ) + + +def test_download_image_simulate(): + args = argparse.Namespace(simulate=True) + avi = AeonViewImages(default_test_path, f"{default_image_domain}.jpg", args) + with mock.patch("aeonview.logging.info") as log: + avi.download_image(Path(tempfile.gettempdir(), "image.jpg")) + assert any("Simulate" in str(call) for call in log.call_args_list) + + +@mock.patch("aeonview.AeonViewHelpers.mkdir_p") +@mock.patch("aeonview.AeonViewImages.download_image") +def test_image_simulation(mock_download_image, mock_mkdir_p): + args = mock.MagicMock() + args.simulate = True + args.date = "2025-04-10 12:30:45" + avi = AeonViewImages(default_test_path, f"{default_image_domain}.jpg", args) + with mock.patch("aeonview.logging.info") as log: + avi.get_current_image() + mock_mkdir_p.assert_not_called() + mock_download_image.assert_not_called() + assert any( + "Saving image to" in str(call) for call in log.call_args_list + ) + + +# --------------------------------------------------------------------------- +# AeonViewVideos tests +# --------------------------------------------------------------------------- + + @mock.patch("subprocess.run") def test_generate_daily_video(mock_subprocess_run): with tempfile.TemporaryDirectory() as tmp: project_path = Path(tmp).resolve() - args = argparse.Namespace( - simulate=False, fps=10, day="01", month="04", year="2025" - ) + args = make_video_args() avv = AeonViewVideos(project_path, args) # Create input directory so the existence check passes (project_path / "img" / "2025-04" / "01").mkdir(parents=True) @@ -199,30 +399,173 @@ def test_generate_daily_video(mock_subprocess_run): @mock.patch("aeonview.AeonViewHelpers.mkdir_p") def test_generate_daily_video_simulate(mock_mkdir_p): - args = argparse.Namespace( - simulate=True, fps=10, day="01", month="04", year="2025" - ) + args = make_video_args(simulate=True) avv = AeonViewVideos(default_test_path, args) avv.generate_daily_video() mock_mkdir_p.assert_not_called() -def test_generate_monthly_video_not_implemented(): - args = argparse.Namespace( - simulate=False, fps=10, day="01", month="04", year="2025" - ) - avv = AeonViewVideos(default_test_path, args) - with pytest.raises(NotImplementedError): - avv.generate_monthly_video(Path(tempfile.gettempdir())) +def test_generate_daily_video_missing_input_dir(): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + args = make_video_args() + avv = AeonViewVideos(project_path, args) + with expect_error_exit(): + avv.generate_daily_video() -def test_generate_yearly_video_not_implemented(): - args = argparse.Namespace( - simulate=False, fps=10, day="01", month="04", year="2025" - ) +@mock.patch("aeonview.AeonViewHelpers.mkdir_p") +@mock.patch("subprocess.run") +def test_video_simulation(mock_subprocess_run, mock_mkdir_p): + args = mock.MagicMock() + args.simulate = True + args.fps = 10 + args.day = "01" + args.month = "01" + args.year = "2023" avv = AeonViewVideos(default_test_path, args) - with pytest.raises(NotImplementedError): - avv.generate_yearly_video(Path(tempfile.gettempdir())) + with mock.patch("aeonview.logging.info") as log: + avv.generate_daily_video() + mock_mkdir_p.assert_not_called() + mock_subprocess_run.assert_not_called() + assert any( + "Generating video from" in str(call) for call in log.call_args_list + ) + + +# --- Monthly video tests --- + + +@mock.patch("subprocess.run") +def test_generate_monthly_video(mock_subprocess_run): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + vid_dir = project_path / "vid" / "2025-04" + vid_dir.mkdir(parents=True) + # Create fake daily videos + for day in ["01", "02", "03"]: + (vid_dir / f"{day}.mp4").touch() + args = make_video_args() + avv = AeonViewVideos(project_path, args) + avv.generate_monthly_video() + mock_subprocess_run.assert_called_once() + call_cmd = mock_subprocess_run.call_args[0][0] + assert "concat" in call_cmd + assert str(vid_dir / "2025-04.mp4") == call_cmd[-1] + + +@mock.patch("subprocess.run") +def test_generate_monthly_video_excludes_output_file(mock_subprocess_run): + """Verify that the monthly output file is excluded from the input list on re-runs.""" + captured_content = {} + + def capture_concat(cmd, **_kwargs): + idx = cmd.index("-i") + 1 + captured_content["text"] = Path(cmd[idx]).read_text(encoding="utf-8") + + mock_subprocess_run.side_effect = capture_concat + + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + vid_dir = project_path / "vid" / "2025-04" + vid_dir.mkdir(parents=True) + # Create daily videos plus a leftover monthly output + for day in ["01", "02"]: + (vid_dir / f"{day}.mp4").touch() + (vid_dir / "2025-04.mp4").touch() # leftover from previous run + args = make_video_args() + avv = AeonViewVideos(project_path, args) + avv.generate_monthly_video() + mock_subprocess_run.assert_called_once() + content = captured_content["text"] + assert "2025-04.mp4" not in content + assert "01.mp4" in content + assert "02.mp4" in content + + +def test_generate_monthly_video_simulate(): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + vid_dir = project_path / "vid" / "2025-04" + vid_dir.mkdir(parents=True) + (vid_dir / "01.mp4").touch() + args = make_video_args(simulate=True) + avv = AeonViewVideos(project_path, args) + with mock.patch("subprocess.run") as mock_run: + avv.generate_monthly_video() + mock_run.assert_not_called() + + +def test_generate_monthly_video_no_input_dir(): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + # Don't create the vid directory + args = make_video_args() + avv = AeonViewVideos(project_path, args) + with expect_error_exit(): + avv.generate_monthly_video() + + +def test_generate_monthly_video_no_mp4_files(): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + vid_dir = project_path / "vid" / "2025-04" + vid_dir.mkdir(parents=True) + # No mp4 files in directory + args = make_video_args() + avv = AeonViewVideos(project_path, args) + with expect_error_exit(): + avv.generate_monthly_video() + + +# --- Yearly video tests --- + + +@mock.patch("subprocess.run") +def test_generate_yearly_video(mock_subprocess_run): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + # Create monthly video files in their directories + for month in ["01", "02", "03"]: + month_dir = project_path / "vid" / f"2025-{month}" + month_dir.mkdir(parents=True) + (month_dir / f"2025-{month}.mp4").touch() + args = make_video_args(month="01", year="2025") + avv = AeonViewVideos(project_path, args) + avv.generate_yearly_video() + mock_subprocess_run.assert_called_once() + call_cmd = mock_subprocess_run.call_args[0][0] + assert "concat" in call_cmd + output_dir = project_path / "vid" / "2025" + assert str(output_dir / "2025.mp4") == call_cmd[-1] + + +def test_generate_yearly_video_simulate(): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + month_dir = project_path / "vid" / "2025-01" + month_dir.mkdir(parents=True) + (month_dir / "2025-01.mp4").touch() + args = make_video_args(simulate=True, month="01", year="2025") + avv = AeonViewVideos(project_path, args) + with mock.patch("subprocess.run") as mock_run: + avv.generate_yearly_video() + mock_run.assert_not_called() + + +def test_generate_yearly_video_no_monthly_videos(): + with tempfile.TemporaryDirectory() as tmp: + project_path = Path(tmp).resolve() + (project_path / "vid").mkdir(parents=True) + args = make_video_args(month="01", year="2025") + avv = AeonViewVideos(project_path, args) + with expect_error_exit(): + avv.generate_yearly_video() + + +# --------------------------------------------------------------------------- +# Argument parsing tests +# --------------------------------------------------------------------------- @mock.patch( @@ -291,39 +634,9 @@ def test_parse_arguments_defaults(): assert not args.verbose -@mock.patch("aeonview.AeonViewHelpers.mkdir_p") -@mock.patch("aeonview.AeonViewImages.download_image") -def test_image_simulation(mock_download_image, mock_mkdir_p): - args = mock.MagicMock() - args.simulate = True - args.date = "2025-04-10 12:30:45" - avi = AeonViewImages(default_test_path, f"{default_image_domain}.jpg", args) - with mock.patch("aeonview.logging.info") as log: - avi.get_current_image() - mock_mkdir_p.assert_not_called() - mock_download_image.assert_not_called() - assert any( - "Saving image to" in str(call) for call in log.call_args_list - ) - - -@mock.patch("aeonview.AeonViewHelpers.mkdir_p") -@mock.patch("subprocess.run") -def test_video_simulation(mock_subprocess_run, mock_mkdir_p): - args = mock.MagicMock() - args.simulate = True - args.fps = 10 - args.day = "01" - args.month = "01" - args.year = "2023" - avv = AeonViewVideos(default_test_path, args) - with mock.patch("aeonview.logging.info") as log: - avv.generate_daily_video() - mock_mkdir_p.assert_not_called() - mock_subprocess_run.assert_not_called() - assert any( - "Generating video from" in str(call) for call in log.call_args_list - ) +# --------------------------------------------------------------------------- +# Logger tests +# --------------------------------------------------------------------------- @mock.patch("logging.basicConfig") @@ -340,3 +653,256 @@ def test_setup_logger_non_verbose(mock_basic_config): mock_basic_config.assert_called_once_with( level=logging.INFO, format="[%(levelname)s] %(message)s" ) + + +# --------------------------------------------------------------------------- +# AeonViewApp tests +# --------------------------------------------------------------------------- + + +@mock.patch( + "sys.argv", + ["aeonview", "--mode", "image", "--url", "https://example.com/image.jpg"], +) +def test_app_init(): + app = AeonViewApp() + assert app.args.mode == "image" + assert app.args.url == "https://example.com/image.jpg" + + +@mock.patch( + "sys.argv", + ["aeonview", "--mode", "image", "--url", "https://example.com/image.jpg"], +) +@mock.patch("aeonview.AeonViewApp.process_image") +def test_app_run_image_mode(mock_process): + app = AeonViewApp() + app.run() + mock_process.assert_called_once() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "video", + "--project", + "myproj", + ], +) +@mock.patch("aeonview.AeonViewApp.process_video") +def test_app_run_video_mode(mock_process): + app = AeonViewApp() + app.run() + mock_process.assert_called_once() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "image", + "--simulate", + "--url", + "https://example.com/image.jpg", + ], +) +def test_app_run_simulate_logs(): + app = AeonViewApp() + with ( + mock.patch("aeonview.AeonViewApp.process_image"), + mock.patch("aeonview.logging.info") as log, + ): + app.run() + assert any("Simulation" in str(call) for call in log.call_args_list) + + +@mock.patch("sys.argv", ["aeonview", "--mode", "image"]) +def test_app_process_image_no_url(): + app = AeonViewApp() + with expect_error_exit(): + app.process_image() + + +@mock.patch("sys.argv", ["aeonview", "--mode", "image", "--url", "not-http"]) +def test_app_process_image_invalid_url(): + app = AeonViewApp() + with expect_error_exit(): + app.process_image() + + +@mock.patch( + "sys.argv", + ["aeonview", "--mode", "image", "--url", "https://example.com/image.jpg"], +) +@mock.patch("aeonview.AeonViewImages.get_current_image") +def test_app_process_image_default_project_hash(mock_get_image): + app = AeonViewApp() + app.process_image() + mock_get_image.assert_called_once() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "image", + "--url", + "https://example.com/image.jpg", + "--project", + "myproject", + ], +) +@mock.patch("aeonview.AeonViewImages.get_current_image") +def test_app_process_image_named_project(mock_get_image): + app = AeonViewApp() + app.process_image() + mock_get_image.assert_called_once() + + +@mock.patch( + "sys.argv", + ["aeonview", "--mode", "video", "--project", ""], +) +def test_app_process_video_no_project(): + app = AeonViewApp() + app.args.project = "" + with expect_error_exit(): + app.process_video() + + +@mock.patch( + "sys.argv", + ["aeonview", "--mode", "video", "--project", "nonexistent"], +) +def test_app_process_video_missing_path(): + app = AeonViewApp() + with expect_error_exit(): + app.process_video() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "video", + "--project", + "proj", + "--generate", + "2025-04-10", + ], +) +@mock.patch("aeonview.AeonViewVideos.generate_daily_video") +def test_app_process_video_with_generate_date(mock_gen): + with tempfile.TemporaryDirectory() as tmp: + app, _ = make_app_with_project(tmp) + app.process_video() + mock_gen.assert_called_once() + + +@mock.patch( + "sys.argv", + ["aeonview", "--mode", "video", "--project", "proj"], +) +@mock.patch("aeonview.AeonViewVideos.generate_daily_video") +def test_app_process_video_default_date(mock_gen): + with tempfile.TemporaryDirectory() as tmp: + app, _ = make_app_with_project(tmp) + app.process_video() + mock_gen.assert_called_once() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "video", + "--project", + "proj", + "--generate", + "invalid-date", + ], +) +def test_app_process_video_invalid_date(): + with tempfile.TemporaryDirectory() as tmp: + app, _ = make_app_with_project(tmp) + with expect_error_exit(): + app.process_video() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "video", + "--project", + "proj", + "--timeframe", + "monthly", + ], +) +@mock.patch("aeonview.AeonViewVideos.generate_monthly_video") +def test_app_process_video_monthly(mock_gen): + with tempfile.TemporaryDirectory() as tmp: + app, _ = make_app_with_project(tmp) + app.process_video() + mock_gen.assert_called_once() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "video", + "--project", + "proj", + "--timeframe", + "yearly", + ], +) +@mock.patch("aeonview.AeonViewVideos.generate_yearly_video") +def test_app_process_video_yearly(mock_gen): + with tempfile.TemporaryDirectory() as tmp: + app, _ = make_app_with_project(tmp) + app.process_video() + mock_gen.assert_called_once() + + +@mock.patch("sys.argv", ["aeonview"]) +def test_app_init_no_args(): + with ( + mock.patch( + "aeonview.AeonViewHelpers.parse_arguments", + return_value=(None, argparse.ArgumentParser()), + ), + pytest.raises(SystemExit), + ): + AeonViewApp() + + +@mock.patch( + "sys.argv", + [ + "aeonview", + "--mode", + "video", + "--project", + "proj", + "--generate", + "2025-04-10", + ], +) +def test_app_process_video_invalid_project_type(): + with tempfile.TemporaryDirectory() as tmp: + app, _ = make_app_with_project(tmp) + # noinspection PyTypeChecker + app.args.project = 12345 # pyright: ignore [reportAttributeAccessIssue] + with expect_error_exit(): + app.process_video()