diff --git a/.editorconfig b/.editorconfig index 58708d8..9d0107b 100644 --- a/.editorconfig +++ b/.editorconfig @@ -3,17 +3,17 @@ # top-most EditorConfig file root = true -# Unix-style newlines with a newline ending every file [*] -end_of_line = lf -insert_final_newline = true - -# Matches multiple files with brace expansion notation -# Set default charset -[*.{js,py}] charset = utf-8 - -# 4 space indentation -[*.py] -indent_style = tab +end_of_line = lf +max_line_length = 100 +insert_final_newline = true +indent_style = space indent_size = 2 +trim_trailing_whitespace = true + +[*.py] +indent_size = 4 + +[Makefile] +indent_style = tab diff --git a/.github/CODE_OF_CONDUCT.md b/.github/CODE_OF_CONDUCT.md new file mode 100644 index 0000000..2a55daa --- /dev/null +++ b/.github/CODE_OF_CONDUCT.md @@ -0,0 +1,57 @@ +# Contributor Covenant Code of Conduct + +## Our Pledge + +We as members, contributors, and leaders pledge to make participation +in our community a harassment-free experience for everyone, regardless +of age, body size, visible or invisible disability, ethnicity, sex +characteristics, gender identity and expression, level of experience, +education, socio-economic status, nationality, personal appearance, +race, caste, color, religion, or sexual identity and orientation. + +We pledge to act and interact in ways that contribute to an open, +welcoming, diverse, inclusive, and healthy community. + +## Our Standards + +Examples of behavior that contributes to a positive environment: +- Demonstrating empathy and kindness +- Being respectful of differing opinions +- Gracefully accepting constructive feedback +- Focusing on what is best for the community + +Examples of unacceptable behavior: +- The use of sexualized language or imagery +- Trolling, insulting or derogatory comments +- Harassment of any kind +- Publishing others’ private information + +## Enforcement Responsibilities + +Community leaders are responsible for clarifying and enforcing our +standards and will take appropriate and fair corrective action in +response to any behavior they deem inappropriate. + +## Scope + +This Code of Conduct applies within all community spaces and also +applies when an individual is officially representing the project. + +## Enforcement + +Instances of abusive, harassing, or otherwise unacceptable behavior +may be reported to the project owner @ivuorinen. + +All complaints will be reviewed and investigated and will result in a +response that is deemed necessary and appropriate to the circumstances. + +## Attribution + +This Code of Conduct is adapted from the [Contributor Covenant][cc], +version 2.1. + +[cc]: https://www.contributor-covenant.org/version/2/1/code_of_conduct/ + + diff --git a/.github/CONTRIBUTING.md b/.github/CONTRIBUTING.md new file mode 100644 index 0000000..96aabb9 --- /dev/null +++ b/.github/CONTRIBUTING.md @@ -0,0 +1,85 @@ +# Contributing to aeonview + +Thanks for your interest in contributing to **aeonview**! This guide will help you get started. + +## πŸ›  Project Setup + +1. Clone the repository: + ```bash + git clone https://github.com/your-username/aeonview.git + cd aeonview + ``` + +2. Set up your environment: + ```bash + python3 -m venv venv + source venv/bin/activate + pip install -r dev-requirements.txt + ``` + +3. Install pre-commit hooks: + ```bash + pre-commit install + ``` + +## βœ… Development Workflow + +- Make sure your changes are **well-tested**. +- Use `make check` to run linting and tests. +- Follow the existing coding style (Ruff will enforce it). +- All new features must include documentation. + +## πŸ§ͺ Running Tests + +```bash +make test +``` + +## 🧹 Formatting & Linting + +```bash +make format # auto-format code +make lint # check for lint errors +``` + +## βœ… Submitting a Pull Request + +1. Create a feature branch: + ```bash + git checkout -b feature/my-new-feature + ``` + +2. Commit your changes: + ```bash + git add . + git commit -m "feat: add support for X" + ``` + +3. Push and open a pull request: + ```bash + git push origin feature/my-new-feature + ``` + +4. Follow the PR template and link any relevant issues. + +## πŸ“‹ Commit Message Guidelines + +Use [Conventional Commits](https://www.conventionalcommits.org/en/v1.0.0/): + +Examples: +- `feat: add monthly video generation` +- `fix: handle invalid date error` +- `docs: update usage instructions` + +## πŸ™ Code of Conduct + +We expect contributors to follow our [Code of Conduct](CODE_OF_CONDUCT.md). + +## Questions? +Feel free to open an issue or start a discussion! + +Thanks for helping make Aeonview better πŸ’œ + + diff --git a/.github/renovate.json b/.github/renovate.json new file mode 100644 index 0000000..336a46f --- /dev/null +++ b/.github/renovate.json @@ -0,0 +1,16 @@ +{ + "$schema": "https://docs.renovatebot.com/renovate-schema.json", + "extends": [ + "github>ivuorinen/renovate-config" + ], + "packageRules": [ + { + "managers": [ + "github-actions" + ], + "schedule": [ + "daily" + ] + } + ] +} diff --git a/.github/workflows/pr-lint.yml b/.github/workflows/pr-lint.yml new file mode 100644 index 0000000..436386e --- /dev/null +++ b/.github/workflows/pr-lint.yml @@ -0,0 +1,29 @@ +--- +# yaml-language-server: $schema=https://json.schemastore.org/github-workflow.json +name: PR Lint + +# yamllint disable-line rule:truthy +on: + push: + branches: [master, main] + pull_request: + branches: [master, main] + +permissions: read-all + +env: + TRIVY_SEVERITY: CRITICAL,HIGH + DISABLE_LINTERS: GO_GOLANGCI_LINT + +jobs: + Linter: + name: PR Lint + runs-on: ubuntu-latest + permissions: + contents: write # only for delete-branch option + issues: write + pull-requests: write + statuses: write + + steps: + - uses: ivuorinen/actions/pr-lint@312c00f77fbd73948441c8b004607e098e40c97f # 25.4.8 diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml new file mode 100644 index 0000000..0912983 --- /dev/null +++ b/.github/workflows/python-tests.yml @@ -0,0 +1,35 @@ +name: Python tests + +on: + push: + branches: [ main, master ] + pull_request: + branches: [ main, master ] + +permissions: read-all + +jobs: + test: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + + - name: Set up Python + uses: actions/setup-python@65d7f2d534ac1bc67fcd62888c5f4f3d2cb2b236 # v4.7.1 + + - name: Install dependencies + shell: bash + run: | + python -m pip install --upgrade pip + pip install pytest pytest-cov ruff + + - name: Run Ruff linting + shell: bash + run: ruff check . + + - name: Run tests with coverage + shell: bash + run: | + pytest --cov=aeonview --cov-report=term-missing diff --git a/.gitignore b/.gitignore index fce81b6..7a4e45f 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,8 @@ projects/* !projects/.gitkeep .vscode .coverage +.pytest_cache +__pycache__ +.coverage +.ruff_cache +.venv diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..c9b0216 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,34 @@ +repos: +- repo: https://github.com/astral-sh/ruff-pre-commit + rev: "v0.11.4" + hooks: + - id: ruff + args: ["--fix"] + - id: ruff-format +- repo: https://github.com/pre-commit/pre-commit-hooks + rev: v5.0.0 + hooks: + - id: check-ast + - id: check-builtin-literals + - id: check-docstring-first + - id: check-executables-have-shebangs + - id: check-json + - id: check-merge-conflict + - id: check-shebang-scripts-are-executable + - id: check-toml + - id: check-yaml + - id: debug-statements + - id: detect-aws-credentials + - id: double-quote-string-fixer + - id: end-of-file-fixer + - id: mixed-line-ending + - id: name-tests-test + - id: no-commit-to-branch + - id: requirements-txt-fixer + - id: trailing-whitespace +- repo: https://github.com/christophmeissner/pytest-pre-commit + rev: 1.0.0 + hooks: + - id: pytest + pass_filenames: false + always_run: true diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..3e388a4 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.13.2 diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..560b2f4 --- /dev/null +++ b/Makefile @@ -0,0 +1,19 @@ +.PHONY: install format lint test check clean + +install: + @pip install -r requirements.txt + +format: + @ruff format . + +lint: + @ruff check . + +test: + @pytest --cov=aeonview --cov-report=term-missing + +check: lint test + +clean: + @find . -type d -name '__pycache__' -exec rm -r {} + + @rm -rf .pytest_cache .coverage htmlcov diff --git a/README.md b/README.md index b8fa28c..112fafa 100644 --- a/README.md +++ b/README.md @@ -1,20 +1,96 @@ - _) - _` | -_) _ \ \ \ \ / | -_) \ \ \ / - \__,_| \___| \___/ _| _| \_/ _| \___| \_/\_/ - aeonview - a simple timelapse tool +``` + _) + _` | -_) _ \ \ \ \ / | -_) \ \ \ / + \__,_| \___| \___/ _| _| \_/ _| \___| \_/\_/ + aeonview - a simple timelapse tool +``` -aeonview is a tool for automagical timelapse-video generation. -it works as a glue between different linux programs to produce -videos of elapsing time. works best with webcam-images from the net. +**aeonview** is a Python-based tool for generating timelapse videos +from webcam images using `ffmpeg`. It supports automated image +downloading, video stitching, and is fully scriptable via CLI. +Includes developer tooling and tests. -sample: - http://www.youtube.com/watch?v=SnywvnjHpUk +[![CI][ci-b]][ci-l] [![ruff][cc-b]][cc-l] [![MIT][lm-b]][lm-l] +Low quality sample: [aeonview 2min preview/Tampere Jan. 2008][sample] -Needed components: +## Features -* Python -* curl -* mencoder -* lots of harddrive space -* cron +- Timelapse image capture (`--mode image`) +- Video generation (`--mode video`) +- Support for daily, monthly, yearly video runs *(daily implemented)* +- Uses `ffmpeg` and `curl` +- Fully tested with `pytest` +- Linting and formatting via `ruff` +- Pre-commit hooks and CI-ready + +## Requirements + +- Python 3.11+ +- `ffmpeg` and `curl` (system tools) +- lots of hard drive space +- Optional: `pyenv` for managing Python versions + (see `pyenv_requirements`) + +## Installation + +```bash +# Clone the repo +git clone https://github.com/ivuorinen/aeonview.git +cd aeonview + +# Install dependencies +python3 -m venv venv +source venv/bin/activate +pip install -r requirements.txt + +# Install pre-commit hooks +pre-commit install +``` + +## Usage + +```bash +# Capture an image +python aeonview.py \ + --mode image \ + --project example \ + --url "https://example.com/webcam.jpg" + +# Generate a video from yesterday's images +python aeonview.py --mode video --project example +``` + +## Development + +```bash +# Format code +make format + +# Lint code +make lint + +# Run tests +make test +``` + +## System Setup for ffmpeg + +```bash +sudo apt update +sudo apt install ffmpeg curl +``` + +## License + +MIT License Β© 2025 Ismo Vuorinen + +[ci-b]: https://github.com/ivuorinen/aeonview/actions/workflows/python-tests.yml/badge.svg +[ci-l]: https://github.com/ivuorinen/aeonview/actions/workflows/python-tests.yml +[cc-b]: https://img.shields.io/badge/code%20style-ruff-blueviolet +[cc-l]: https://github.com/astral-sh/ruff +[lm-b]: https://img.shields.io/badge/License-MIT-yellow.svg +[lm-l]: https://opensource.org/licenses/MIT +[sample]: https://www.youtube.com/watch?v=SnywvnjHpUk + + diff --git a/aeonview.py b/aeonview.py old mode 100755 new mode 100644 index 5cd7f70..3e211dd --- a/aeonview.py +++ b/aeonview.py @@ -1,221 +1,460 @@ -import sys, time, datetime, os, optparse, errno, re +import argparse +import hashlib +import logging +import os +import subprocess +import sys +from datetime import datetime, timedelta +from pathlib import Path -def aeonview(argv): - """ - aeonview is a tool for automagical timelapse-video generation. - it works as a glue between different linux programs to produce - videos of elapsing time. works best with webcam-images from the net. - """ - version = "0.1.8" - - parser = optparse.OptionParser( - usage="Usage: %prog [options]", - description="aeonview for timelapses", - version="%prog v"+version - ) - - basicopts = optparse.OptionGroup( - parser, "Basic settings", "These effect in both modes." - ) - - basicopts.add_option("-m", "--mode", - default="image", - help="run mode: image or video [default: %default]") - - basicopts.add_option("-p", "--project", - help="Project name, used as directory name. " - "Defaults to 5 characters from md5 hash of the webcam url.", - type="string") - - basicopts.add_option("-d", "--dest", - help="Start of the destination. [default: %default]", - type="string", - default=os.getcwdu()+"/projects", - dest="path") - - basicopts.add_option("--mencoder", - help="Path to mencoder binary. [default: %default]", - type="string", - default=os.getcwdu()+'/mencoder') - - parser.add_option_group(basicopts) - - # When mode is: image - imageopts = optparse.OptionGroup( - parser, "Options for --mode: image", "When we are gathering images.") - - imageopts.add_option("--url", help="Webcam URL", type="string") - - parser.add_option_group(imageopts) - - # When mode is: video - videoopts = optparse.OptionGroup(parser, - "Options for --mode: video", "When we are making movies.") - - videoopts.add_option("--videorun", - default="daily", - help="Video to process: daily, monthly or yearly [default: %default]", - type="string") - - videoopts.add_option('--generate', - help="Date to video. Format: YYYY-MM-DD. " - "Default is calculated yesterday, currently %default", - type="string", - default=datetime.date.today()-datetime.timedelta(1)) - - # TODO: mode for monthly videos - #videoopts.add_option("--gen-month", - # help="Month to video. Format: YYYY-MM. " - # "Default is last month, currently %default", - # type="string", - # default=datetime.date.today()-datetime.timedelta(30)) - - videoopts.add_option("--fps", - default="10", - help="Frames per second, numeric [default: %default]", - type="int") - - parser.add_option_group(videoopts) - - parser.add_option("-v", help="Verbose", action="store_true", dest="verbose", default=False) - parser.add_option("-q", help="Quiet", action="store_false", dest="verbose", default=True) - - parser.add_option("-s", "--simulate", - help="Demostrates what will happen (good for checking your settings and destinations)", - default=False, - action="store_true") - - (options, args) = parser.parse_args(argv[1:]) - - if options.simulate == True: - print - print "--- Starting simulation, just echoing steps using your parameters." - print - print "(!) You are running aeonview from", os.getcwdu() - - if options.mode == 'image': - # We are now in the gathering mode. - - if options.url == None and options.simulate == True: - options.url = "http://example.com/webcam.jpg" - print "(!) Simulation: Using " + options.url + " as webcam url" - - if options.url == None: - print "(!) Need a webcam url, not gonna rock before that!" - print - parser.print_help() - sys.exit(-1) - - if options.project == None: - import hashlib # not needed before - m = hashlib.md5(options.url).hexdigest() - options.project = m[:5] # 5 first characters of md5-hash - if options.verbose == True or options.simulate == True: - print "(!) No project defined, using part of md5-hash of the webcam url:", options.project - - if options.path == None: - if options.verbose == True or options.simulate == True: - print "(!) No destination defined, using:", options.path - else: - if options.verbose == True or options.simulate == True: - print "(!) Using destination:", options.path - - # If you want to change the path structure, here's your chance. - options.imgpath = time.strftime("/img/%Y-%m/%d/") - options.imgname = time.strftime("%H-%M-%S") - - # Let us build the destination path and filename - options.fileext = os.path.splitext(options.url)[1] - options.destdir = options.path + "/" + options.project + options.imgpath - options.destination = options.destdir + options.imgname + options.fileext - getit = options.url + " -o " + options.destination - - # Crude, but works. - if options.simulate == False: - os.system("curl --create-dirs --silent %s" % getit) - else: - print "(!) Simulation: Making path:", options.destdir - print "(!) Simulation: curl (--create-dirs and --silent)", getit - - elif options.mode == "video": - # We are now in the video producing mode - - vid_extension = ".avi" - #m = os.getcwd() + "/mencoder" - m = options.mencoder - mencoder = m + " -really-quiet -mf fps="+ str(options.fps) +" -nosound -ovc lavc -lavcopts vcodec=mpeg4" - - if options.project == None: - print "(!) No project defined, please specify what project you are working on." - print - parser.print_help() - sys.exit(-1) - - if options.videorun == "daily": - vid_date = str(options.generate).split("-") - year = vid_date[0] - month = vid_date[1] - day = vid_date[2] - - if check_date(int(year), int(month), int(day)): - proj_dir = options.path + "/" + options.project - video_dir = proj_dir + "/img/" + year + "-" + month + "/" + day + "/*" - video_out_dir = proj_dir + "/vid/" + year + "-" + month + "/" - video_out_day = video_out_dir + day + vid_extension - mfdir = os.path.dirname(os.path.realpath(video_dir)) - command = mencoder + " -o " + video_out_day + " 'mf://" + mfdir + "/*'" - - if options.simulate == False: - mkdir_p( video_out_dir ) - os.system(command) - else: - print "(!) Video dir to process:", video_dir - print "(!) Video output-file:", video_out_day - print "(!) Made directory structure:", video_out_dir - print "(!) Command to run", command - - else: - print "(!) Error: check your date. Value provided:", options.generate - - elif options.videorun == "monthly": - print "Monthly: TODO" - # TODO Monthly script. Joins daily movies of that month. - - elif options.videorun == "yearly": - print "Yearly: TODO" - # TODO Yearly script. Joins monthly movies together. - - else: - print "(!) What? Please choose between -r daily/montly/yearly" - - else: - parser.print_help() - sys.exit(-1) +import requests -# http://stackoverflow.com/questions/600268/mkdir-p-functionality-in-python/600612#600612 -def mkdir_p(path): - try: - os.makedirs(path) - except OSError as exc: # Python >2.5 - if exc.errno == errno.EEXIST: - pass - else: raise +# Define messages for logging +# These messages are used for logging purposes and can be customized as needed. +class AeonViewMessages: + INVALID_URL = 'Invalid URL provided.' + INVALID_DATE = 'Invalid date format provided.' + DOWNLOAD_SUCCESS = 'Image downloaded successfully.' + DOWNLOAD_FAILURE = 'Failed to download image.' + VIDEO_GENERATION_SUCCESS = 'Video generated successfully.' + VIDEO_GENERATION_FAILURE = 'Failed to generate video.' + INVALID_IMAGE_FORMAT = 'Invalid image format provided.' + INVALID_IMAGE_EXTENSION = 'Invalid image extension provided.' -# Modified http://markmail.org/message/k2pxsle2lslrmnut -def check_date(year, month, day): - tup1 = (year, month, day, 0,0,0,0,0,0) - try: - date = time.mktime (tup1) - tup2 = time.localtime (date) - if tup1[:2] != tup2[:2]: - return False - else: - return True - except OverflowError: - return False +class AeonViewImages: + """ + Class to handle image download and saving. + + This class is responsible for downloading images from a URL and saving them + to a specified directory. + """ + + 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 + """ + self.project_path = project_path or None + self.url = url or None + self.args = args or {} + + def get_image_paths( + self, url: str | None, destination_base: Path | None, date: datetime + ) -> 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 + """ + if url is None or not isinstance(url, str): + logging.error(AeonViewMessages.INVALID_URL) + sys.exit(1) + if not isinstance(date, datetime): + logging.error(AeonViewMessages.INVALID_DATE) + sys.exit(1) + if not url.startswith('http'): + logging.error(AeonViewMessages.INVALID_URL) + sys.exit(1) + if not url.endswith(('.jpg', '.jpeg', '.png', '.gif', '.webp')): + logging.error(AeonViewMessages.INVALID_IMAGE_FORMAT) + sys.exit(1) + + if destination_base is None: + logging.error('No destination base path provided.') + sys.exit(1) + if not isinstance(destination_base, Path): + logging.error('Invalid destination base path.') + sys.exit(1) + + year = date.strftime('%Y') + month = date.strftime('%m') + day = date.strftime('%d') + + year_month = f'{year}-{month}' + + destination = AeonViewHelpers.build_path(destination_base, year_month, day) + file_name = date.strftime('%H-%M-%S') + AeonViewHelpers.get_extension(url) + destination_file = AeonViewHelpers.build_path(destination, file_name) + + if not destination.exists(): + if self.args.simulate: + logging.info(f'Simulate: would create {destination}') + else: + AeonViewHelpers.mkdir_p(destination) + logging.info(f'Creating destination base path: {destination}') + + return { + 'url': url, + 'file': file_name, + 'date': { + 'year': year, + 'month': month, + 'day': day, + 'hour': date.strftime('%H'), + 'minute': date.strftime('%M'), + 'second': date.strftime('%S'), + }, + 'destinations': { + 'base': destination_base, + 'year_month': year_month, + 'day': day, + 'file': destination_file, + }, + } + + def get_current_image(self): + """ + Download the image from the URL and save it to the project directory. + """ + + if self.args.date is not None: + try: + date = datetime.strptime(self.args.date, '%Y-%m-%d %H:%M:%S') + except ValueError: + logging.error(AeonViewMessages.INVALID_DATE) + sys.exit(1) + else: + date = datetime.now() + + img_path = date.strftime('img/%Y-%m/%d') + img_name = date.strftime('%H-%M-%S') + + if self.url is None: + logging.error(AeonViewMessages.INVALID_URL) + sys.exit(1) + + file_ext = AeonViewHelpers.get_extension(self.url) + + dest_dir = AeonViewHelpers.build_path(self.project_path, img_path) + dest_file = AeonViewHelpers.build_path(dest_dir, f'{img_name}{file_ext}') + + logging.info(f'Saving image to {dest_file}') + + if not self.args.simulate: + AeonViewHelpers.mkdir_p(dest_dir) + self.download_image(dest_file) + else: + logging.info(f'Simulate: would create {dest_dir}') + logging.info(f'Simulate: would download {self.url} to {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 + """ + + if self.url is None: + logging.error(AeonViewMessages.INVALID_URL) + sys.exit(1) + + if not isinstance(destination, Path): + logging.error('Invalid destination path.') + sys.exit(1) + + if self.args.simulate is False or self.args.simulate is None: + logging.info(f'Downloading image from {self.url}') + response = requests.get(self.url, stream=True) + if response.status_code == 200: + with open(destination, 'wb') as f: + for chunk in response.iter_content(1024): + f.write(chunk) + logging.info(f'{AeonViewMessages.DOWNLOAD_SUCCESS}: {destination}') + else: + logging.error(f'{AeonViewMessages.DOWNLOAD_FAILURE}: {self.url}') + sys.exit(1) + else: + logging.info(f'Simulate: would download {self.url} to {destination}') + + +class AeonViewVideos: + """ + Class to handle video generation and management. + + This class is responsible for generating daily, monthly, and yearly videos. + It uses ffmpeg for video processing. + """ + + def __init__(self, project_path: Path, args=None): + """ + 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 or {} + + self.args.simulate = args.simulate or False + self.args.fps = args.fps or 10 + + self.day = args.day or None + self.month = args.month or None + self.year = args.year or None + + self.path_images = AeonViewHelpers.build_path(self.project_path, 'img') + self.path_videos = AeonViewHelpers.build_path(self.project_path, 'vid') + + def generate_daily_video(self): + """ + Generate a daily video from images. + """ + + year_month = f'{self.year}-{self.month}' + + input_dir = AeonViewHelpers.build_path(self.path_videos, year_month, self.day) + output_dir = AeonViewHelpers.build_path(self.path_videos, year_month) + output_file = AeonViewHelpers.build_path(output_dir, f'{self.day}.mp4') + ffmpeg_cmd = AeonViewHelpers.generate_ffmpeg_command(input_dir, output_file, self.args.fps) + + logging.info(f'Generating video from {input_dir}') + logging.info(f'Output file will be {output_file}') + + if not self.args.simulate: + logging.info(f'Running ffmpeg command: {" ".join(ffmpeg_cmd)}') + if not os.path.exists(input_dir): + AeonViewHelpers.mkdir_p(output_dir) + subprocess.run(ffmpeg_cmd, check=True) + logging.info(f'{AeonViewMessages.VIDEO_GENERATION_SUCCESS}: {output_file}') + else: + logging.info(f'Simulate: would run {" ".join(ffmpeg_cmd)}') + + def generate_monthly_video(self, output_dir: Path): + """ + 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.') + + def generate_yearly_video(self, output_dir: Path): + """ + Generate a yearly video from images. + :param output_dir: Directory where the video will be saved + :return: None + """ + raise NotImplementedError('Yearly video generation is not implemented.') + + +class AeonViewHelpers: + """ + Helper class for common operations. + """ + + @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 + """ + try: + date = datetime(year, month, day) + return date.year == year and date.month == month and date.day == day + except ValueError: + return False + + @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 + """ + 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 + """ + return Path(base).joinpath(*args or []).resolve() + + @staticmethod + def parse_arguments(): + """Parse command line arguments.""" + + dest_default = str(Path.cwd() / 'projects') + + parser = argparse.ArgumentParser(description='aeonview - timelapse generator using ffmpeg') + parser.add_argument('--mode', choices=['image', 'video'], default='image', help='Run mode') + parser.add_argument('--project', help='Project name', default='default') + parser.add_argument('--dest', default=dest_default, help='Destination root path') + parser.add_argument('--url', help='Webcam URL (required in image mode)') + parser.add_argument('--fps', type=int, default=10, help='Frames per second') + parser.add_argument('--generate', help='Date for video generation (YYYY-MM-DD)') + parser.add_argument('--timeframe', choices=['daily', 'monthly', 'yearly'], default='daily') + parser.add_argument( + '--simulate', action='store_true', help='Simulation mode', default=False + ) + parser.add_argument('--verbose', action='store_true', help='Verbose output', default=False) + args = parser.parse_args() + return args, parser + + @staticmethod + def get_extension(url: str | None) -> str | None: + """ + Get the file extension from the URL. + :return: File extension + """ + if url is None: + logging.error(AeonViewMessages.INVALID_IMAGE_EXTENSION) + return None + + if url.endswith('.png'): + return '.png' + elif url.endswith('.gif'): + return '.gif' + elif url.endswith('.webp'): + return '.webp' + else: + return '.jpg' + + @staticmethod + def setup_logger(verbose: bool): + """ + Set up the logger for the application. + :param verbose: Enable verbose logging if True + :return: None + """ + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig(level=level, format='[%(levelname)s] %(message)s') + + @staticmethod + 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 + """ + return [ + 'ffmpeg', + '-framerate', + str(fps), + '-pattern_type', + 'glob', + '-i', + str(input_dir / '*.{jpg,jpeg,png,gif,webp}'), + '-c:v', + 'libx264', + '-pix_fmt', + 'yuv420p', + str(output_file), + ] + + +class AeonViewApp: + """ + Main application class for AeonView. + """ + + def __init__(self): + self.args, self.parser = AeonViewHelpers.parse_arguments() + AeonViewHelpers.setup_logger(self.args.verbose) + self.base_path = Path(self.args.dest).resolve() + + def run(self): + """ + Execute the application based on the provided arguments. + """ + if self.args.simulate: + logging.info('Simulation mode active. No actions will be executed.') + + if self.args.mode == 'image': + self.process_image() + elif self.args.mode == 'video': + self.process_video() + + def process_image(self): + """ + Process image download and saving based on the provided arguments. + """ + if not self.args.url or self.args.url is None: + logging.error('--url is required in image mode') + self.parser.print_help() + sys.exit(1) + + if not isinstance(self.args.url, str) or not self.args.url.startswith('http'): + logging.error(f'{AeonViewMessages.INVALID_URL}: {self.args.url}') + sys.exit(1) + + url = self.args.url + project = self.args.project or 'default' + + if project == 'default' and url: + project = hashlib.md5(url.encode()).hexdigest()[:5] + + project_path = AeonViewHelpers.build_path(self.base_path, project) + avi = AeonViewImages(project_path, url, self.args) + avi.get_current_image() + + def process_video(self): + """ + Process video generation based on the provided arguments. + """ + if not self.args.project: + logging.error('--project is required in video mode') + self.parser.print_help() + sys.exit(1) + + if not isinstance(self.args.project, str): + logging.error(f'Invalid project name: {self.args.project}') + sys.exit(1) + + project_path = AeonViewHelpers.build_path(self.base_path, self.args.project) + + if not os.path.exists(project_path): + logging.error(f'Project path {project_path} does not exist.') + sys.exit(1) + + try: + generate_date = ( + datetime.strptime(self.args.generate, '%Y-%m-%d') + if self.args.generate + else datetime.today() - timedelta(days=1) + ) + except ValueError: + logging.error(AeonViewMessages.INVALID_DATE) + sys.exit(1) + + year = generate_date.strftime('%Y') + month = generate_date.strftime('%m') + day = generate_date.strftime('%d') + + self.args.day = day + self.args.month = month + self.args.year = year + + if not AeonViewHelpers.check_date(int(year), int(month), int(day)): + logging.error(f'Invalid date: {year}-{month}-{day}') + sys.exit(1) + + avm = AeonViewVideos(project_path, self.args) + + 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) + elif self.args.timeframe == 'yearly': + output_dir = AeonViewHelpers.build_path(project_path, 'vid', year) + avm.generate_yearly_video(output_dir) if __name__ == '__main__': - aeonview(sys.argv) + app = AeonViewApp() + app.run() + +# vim: set tw=100 fo=cqt wm=0 et: diff --git a/aeonview_test.py b/aeonview_test.py new file mode 100644 index 0000000..5f0abb7 --- /dev/null +++ b/aeonview_test.py @@ -0,0 +1,318 @@ +import argparse +import logging +import subprocess +import tempfile +from pathlib import Path +from unittest import TestCase, mock +from datetime import datetime + +import pytest + +from aeonview import AeonViewHelpers, AeonViewImages, AeonViewVideos, AeonViewMessages + +# Define values used in the tests +default_dest = str(Path.cwd() / 'projects') +default_project = 'default' +default_fps = 10 +default_timeframe = 'daily' +default_simulate = False +default_verbose = False +default_image_domain = 'https://example.com/image' +default_test_path = Path('/tmp/test_project').resolve() + + +# Define the Helpers class with methods to be tested +class TestHelpers(TestCase): + def test_check_date_valid(self): + self.assertTrue(AeonViewHelpers.check_date(2023, 12, 31)) + + def test_check_date_invalid(self): + self.assertFalse(AeonViewHelpers.check_date(2023, 2, 30)) + + def test_mkdir_p_creates_directory(self): + with tempfile.TemporaryDirectory() as tmp: + test_path = Path(tmp) / 'a' / 'b' / 'c' + AeonViewHelpers.mkdir_p(test_path) + self.assertTrue(test_path.exists()) + self.assertTrue(test_path.is_dir()) + + def test_get_extension_valid(self): + self.assertEqual(AeonViewHelpers.get_extension(f'{default_image_domain}.png'), '.png') + self.assertEqual(AeonViewHelpers.get_extension(f'{default_image_domain}.jpg'), '.jpg') + self.assertEqual(AeonViewHelpers.get_extension(f'{default_image_domain}.gif'), '.gif') + self.assertEqual(AeonViewHelpers.get_extension(f'{default_image_domain}.webp'), '.webp') + + def test_get_extension_invalid(self): + self.assertEqual( + AeonViewHelpers.get_extension(default_image_domain), '.jpg' + ) # Default behavior + self.assertIsNone(AeonViewHelpers.get_extension(None)) + + +class TestFFmpegCommand(TestCase): + def test_generate_ffmpeg_command(self): + input_dir = Path('/tmp/images') + output_file = Path('/tmp/output.mp4') + fps = 24 + cmd = AeonViewHelpers.generate_ffmpeg_command(input_dir, output_file, fps) + self.assertIn('ffmpeg', cmd[0]) + self.assertIn(str(fps), cmd) + self.assertEqual(str(output_file), cmd[-1]) + self.assertIn(str(input_dir / '*.{jpg,jpeg,png,gif,webp}'), cmd) + + def test_generate_ffmpeg_command_output_format(self): + input_dir = Path('/tmp/images') + output_file = Path('/tmp/video.mp4') + cmd = AeonViewHelpers.generate_ffmpeg_command(input_dir, output_file, 30) + self.assertIn('/tmp/images/*.{jpg,jpeg,png,gif,webp}', cmd) + self.assertIn('/tmp/video.mp4', cmd) + self.assertIn('-c:v', cmd) + self.assertIn('libx264', cmd) + self.assertIn('-pix_fmt', cmd) + self.assertIn('yuv420p', cmd) + + @mock.patch('subprocess.run') + def test_simulate_ffmpeg_call(self, mock_run): + input_dir = Path('/tmp/images') + output_file = Path('/tmp/out.mp4') + cmd = AeonViewHelpers.generate_ffmpeg_command(input_dir, output_file, 10) + subprocess.run(cmd) + mock_run.assert_called_once_with(cmd) + + +class TestAeonViewImages(TestCase): + def setUp(self): + self.args = argparse.Namespace() + self.args.simulate = False + self.args.date = '2025-04-10 12:30:45' + self.args.url = f'{default_image_domain}.jpg' + self.args.dest = default_test_path + self.args.project = default_project + self.args.verbose = default_verbose + self.args.fps = default_fps + self.args.timeframe = default_timeframe + self.project_path = default_test_path + self.url = f'{default_image_domain}.jpg' + + def test_get_image_paths_valid(self): + url = f'{default_image_domain}.jpg' + destination_base = default_test_path + date = datetime(2025, 4, 10, 12, 30, 45) + paths = AeonViewImages.get_image_paths(self, url, destination_base, date) + self.assertEqual(paths['url'], url) + self.assertEqual(paths['file'], '12-30-45.jpg') + self.assertEqual( + paths['destinations']['file'], destination_base / '2025-04' / '10' / '12-30-45.jpg' + ) + + def test_get_image_paths_invalid_url(self): + with pytest.raises(SystemExit): + with self.assertLogs(level='ERROR') as log: + AeonViewImages.get_image_paths( + self, 'invalid-url', default_test_path, datetime(2025, 4, 10) + ) + self.assertIn(AeonViewMessages.INVALID_URL, log.output[0]) + + def test_get_image_paths_invalid_date(self): + with pytest.raises(SystemExit): + with self.assertLogs(level='ERROR') as log: + AeonViewImages.get_image_paths( + self, f'{default_image_domain}.jpg', default_test_path, 'invalid-date' + ) + self.assertIn(AeonViewMessages.INVALID_DATE, log.output[0]) + + @mock.patch('aeonview.AeonViewHelpers.mkdir_p') + @mock.patch('aeonview.AeonViewImages.download_image') + def test_get_current_image(self, mock_download_image, mock_mkdir_p): + project_path = default_test_path + url = f'{default_image_domain}.jpg' + args = argparse.Namespace(simulate=False, date='2025-04-10 12:30:45') + avi = AeonViewImages(project_path, url, args) + avi.get_current_image() + mock_mkdir_p.assert_called() + mock_download_image.assert_called() + + @mock.patch('aeonview.requests.get') + def test_download_image_success(self, mock_get): + mock_response = mock.Mock() + mock_response.status_code = 200 + mock_response.iter_content = mock.Mock(return_value=[b'data']) + mock_get.return_value = mock_response + + project_path = default_test_path + url = f'{default_image_domain}.jpg' + args = argparse.Namespace(simulate=False) + avi = AeonViewImages(project_path, url, args) + destination = Path('/tmp/image.jpg') + avi.download_image(destination) + + mock_get.assert_called_once_with(url, stream=True) + self.assertTrue(destination.exists()) + + @mock.patch('aeonview.requests.get') + def test_download_image_failure(self, mock_get): + mock_response = mock.Mock() + mock_response.status_code = 404 + mock_get.return_value = mock_response + + project_path = default_test_path + url = f'{default_image_domain}.jpg' + args = argparse.Namespace(simulate=False) + avi = AeonViewImages(project_path, url, args) + destination = Path('/tmp/image.jpg') + + with pytest.raises(SystemExit): + with self.assertLogs(level='ERROR') as log: + avi.download_image(destination) + self.assertIn(AeonViewMessages.DOWNLOAD_FAILURE, log.output[0]) + + +class TestAeonViewVideos(TestCase): + @mock.patch('aeonview.AeonViewHelpers.mkdir_p') + @mock.patch('subprocess.run') + def test_generate_daily_video(self, mock_subprocess_run, mock_mkdir_p): + project_path = default_test_path + args = argparse.Namespace(simulate=False, fps=10, day='01', month='04', year='2025') + avv = AeonViewVideos(project_path, args) + with self.assertLogs(level='INFO') as log: + avv.generate_daily_video() + expected_message = f'{AeonViewMessages.VIDEO_GENERATION_SUCCESS}: {default_test_path / "vid/2025-04/01.mp4"}' + self.assertIn(expected_message, log.output[-1]) # Ensure it's the last log entry + mock_mkdir_p.assert_called() + mock_subprocess_run.assert_called() + + @mock.patch('aeonview.AeonViewHelpers.mkdir_p') + def test_generate_daily_video_simulate(self, mock_mkdir_p): + project_path = default_test_path + args = argparse.Namespace(simulate=True, fps=10, day='01', month='04', year='2025') + avv = AeonViewVideos(project_path, args) + avv.generate_daily_video() + mock_mkdir_p.assert_not_called() + + def test_generate_monthly_video_not_implemented(self): + project_path = default_test_path + args = argparse.Namespace(simulate=False, fps=10, day='01', month='04', year='2025') + avv = AeonViewVideos(project_path, args) + with pytest.raises(NotImplementedError): + avv.generate_monthly_video(Path('/tmp')) + + def test_generate_yearly_video_not_implemented(self): + project_path = default_test_path + args = argparse.Namespace(simulate=False, fps=10, day='01', month='04', year='2025') + avv = AeonViewVideos(project_path, args) + with pytest.raises(NotImplementedError): + avv.generate_yearly_video(Path('/tmp')) + + +class TestHelpersArguments(TestCase): + def setUp(self): + self.default_dest = str(Path.cwd() / 'projects') + + @mock.patch( + 'sys.argv', ['aeonview.py', '--mode', 'image', '--url', f'{default_image_domain}.jpg'] + ) + def test_parse_arguments_image_mode(self): + args, _ = AeonViewHelpers.parse_arguments() + self.assertEqual(args.mode, 'image') + self.assertEqual(args.url, f'{default_image_domain}.jpg') + self.assertEqual(args.dest, self.default_dest) + + @mock.patch('sys.argv', ['aeonview.py', '--mode', 'video', '--project', f'{default_project}']) + def test_parse_arguments_video_mode(self): + args, _ = AeonViewHelpers.parse_arguments() + self.assertEqual(args.mode, 'video') + self.assertEqual(args.project, f'{default_project}') + self.assertEqual(args.dest, self.default_dest) + + @mock.patch('sys.argv', ['aeonview.py', '--mode', 'image', '--simulate']) + def test_parse_arguments_simulate_mode(self): + args, _ = AeonViewHelpers.parse_arguments() + self.assertEqual(args.mode, 'image') + self.assertTrue(args.simulate) + + @mock.patch('sys.argv', ['aeonview.py', '--mode', 'video', '--fps', '30']) + def test_parse_arguments_fps(self): + args, _ = AeonViewHelpers.parse_arguments() + self.assertEqual(args.mode, 'video') + self.assertEqual(args.project, f'{default_project}') + self.assertEqual(args.dest, self.default_dest) + self.assertEqual(args.fps, 30) + + @mock.patch('sys.argv', ['aeonview.py', '--mode', 'video', '--generate', '2023-10-01']) + def test_parse_arguments_generate_date(self): + args, _ = AeonViewHelpers.parse_arguments() + self.assertEqual(args.mode, 'video') + self.assertEqual(args.generate, '2023-10-01') + + @mock.patch('sys.argv', ['aeonview.py', '--mode', 'image', '--verbose']) + def test_parse_arguments_verbose(self): + args, _ = AeonViewHelpers.parse_arguments() + self.assertEqual(args.mode, 'image') + self.assertTrue(args.verbose) + + @mock.patch('sys.argv', ['aeonview.py']) + def test_parse_arguments_defaults(self): + args, _ = AeonViewHelpers.parse_arguments() + self.assertEqual(args.mode, 'image') + self.assertEqual(args.project, f'{default_project}') + self.assertEqual(args.dest, self.default_dest) + self.assertEqual(args.fps, 10) + self.assertEqual(args.timeframe, 'daily') + self.assertFalse(args.simulate) + self.assertFalse(args.verbose) + + +class TestAeonViewSimulation(TestCase): + @mock.patch('aeonview.AeonViewHelpers.mkdir_p') + @mock.patch('aeonview.AeonViewImages.download_image') + def test_image_simulation(self, mock_download_image, mock_mkdir_p): + args = mock.MagicMock() + args.simulate = True + args.date = '2025-04-10 12:30:45' + + url = f'{default_image_domain}.jpg' + project_path = Path('/tmp/test_project').resolve() + + avi = AeonViewImages(project_path, url, args) + with mock.patch('aeonview.logging.info') as mock_logging: + avi.get_current_image() + mock_mkdir_p.assert_not_called() + mock_download_image.assert_not_called() + mock_logging.assert_any_call( + f'Saving image to {project_path}/img/2025-04/10/12-30-45.jpg' + ) + + @mock.patch('aeonview.AeonViewHelpers.mkdir_p') + @mock.patch('subprocess.run') + def test_video_simulation(self, mock_subprocess_run, mock_mkdir_p): + args = mock.MagicMock() + args.simulate = True + args.fps = 10 + args.day = '01' + args.month = '01' + args.year = '2023' + project_path = Path('/tmp/test_project').resolve() + + avv = AeonViewVideos(project_path, args) + with mock.patch('aeonview.logging.info') as mock_logging: + avv.generate_daily_video() + mock_mkdir_p.assert_not_called() + mock_subprocess_run.assert_not_called() + mock_logging.assert_any_call(f'Generating video from {project_path}/vid/2023-01/01') + + +class TestSetupLogger(TestCase): + @mock.patch('logging.basicConfig') + def test_setup_logger_verbose(self, mock_basic_config): + AeonViewHelpers.setup_logger(verbose=True) + mock_basic_config.assert_called_once_with( + level=logging.DEBUG, format='[%(levelname)s] %(message)s' + ) + + @mock.patch('logging.basicConfig') + def test_setup_logger_non_verbose(self, mock_basic_config): + AeonViewHelpers.setup_logger(verbose=False) + mock_basic_config.assert_called_once_with( + level=logging.INFO, format='[%(levelname)s] %(message)s' + ) + mock_basic_config.reset_mock() diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..b5fc6c1 --- /dev/null +++ b/conftest.py @@ -0,0 +1,9 @@ +# conftest.py +from unittest import mock + +import pytest + + +@pytest.fixture(autouse=True) +def no_network_requests(monkeypatch): + monkeypatch.setattr('subprocess.run', mock.Mock()) diff --git a/example.sh b/example.sh index c0e385c..471411b 100755 --- a/example.sh +++ b/example.sh @@ -1,3 +1,5 @@ #!/bin/sh -python aeonview.py -p abc-la --url http://abclocal.go.com/three/kabc/webcam/web2-1.jpg +python aeonview.py \ + -p abc-la \ + --url http://abclocal.go.com/three/kabc/webcam/web2-1.jpg diff --git a/mencoder b/mencoder deleted file mode 100755 index 9e868fb..0000000 Binary files a/mencoder and /dev/null differ diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..581e602 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,24 @@ +[tool.ruff] +line-length = 100 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "B", "UP", "C4", "T20"] +ignore = ["E501"] + +[tool.ruff.lint.per-file-ignores] +"test_*.py" = ["S101"] + +[tool.ruff.format] +quote-style = "single" +indent-style = "space" + +[tool.pytest.ini_options] +testpaths = ["."] +python_files = ["*_test.py"] + +[tool.pylint.MAIN] +ignore-patterns='^\.#' +ignore-paths='^\.#' +ignore='CVS,.venv' +disable='attribute-defined-outside-init,invalid-name,missing-docstring,protected-access,too-few-public-methods,format' diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ca8dab3 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +# Functional requirements +requests~=2.32.3 + +# Pre-commit hooks +pre-commit>=3.5.0 + +# Testing +pytest>=8.0.0 +pytest-cov>=4.1.0 +# Linting & formatting +ruff>=0.3.3