feat: full upgrade to python3, tests, etc.

This commit is contained in:
2025-04-10 16:08:34 +03:00
parent 6aa17ba957
commit dd9f538bb4
18 changed files with 1200 additions and 240 deletions

665
aeonview.py Executable file → Normal file
View File

@@ -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: