"""convmonitor.py The conversion monitor polls to convert newly approved images. """ import os, logging, json, subprocess, hashlib, re, itertools import numpy as np from .video import * #================================================================ def decode_filename(name): """Extract metadata from the filename if possible. Returns a parameter dictionary, possibly with default values.""" # pattern = "([a-zA-Z_@\.]+?)_([0-9]*\.[0-9]+).png" pattern = "([a-zA-Z_@\.]+?)\u2423([0-9]+).png" # if the name includes the Unicode separator if '\u2423' in name: pattern = "([^\u2423]*)\u2423([^\u2423]*)\u2423([0-9]+).png" match = re.search(pattern, name) if match is None: return {'title' : 'notitle', 'email' : 'anonymous', 'tempo' : 120} else: fields = match.groups() return {'title' : fields[0], 'email' : fields[1], 'tempo' : int(fields[2]) } else: # if the name doesn't include the Unicode separator, just look for a tempo pattern = "(.*)[_=]([0-9]+).png" match = re.search(pattern, name) if match is None: return {'title' : 'notitle', 'email' : 'anonymous', 'tempo' : 120} else: fields = match.groups() return {'title' : fields[0], 'email' : 'anonymous', 'tempo' : int(fields[1]) } #================================================================ def calculate_slate_identifer(filename): # take the first few hex digits of the MD5 hash of the filename hashvalue = hashlib.md5(filename.encode()) prefix = hashvalue.hexdigest()[0:4] # convert to an integer value = int(prefix, 16) # return a five-digit string with zero padding # e.g. hash value 234 will generate '00234' return "%05d" % value #================================================================ class ConversionMonitor(object): def __init__(self, config, gdrive): logging.debug("Entering ConversionMonitor.__init__ and initializing superclasses") super(ConversionMonitor, self).__init__() self.polling_interval = 5.0 self.accepted_path = "2-accepted" # images which have been manually accepted and are ready for conversion self.video_path = "4-new-video" # generated video outputs ready for display self.archive_path = "C-processed-submissions" # images which have already been processed self.preview_path = "6-video-preview" # post-processed video more suited to online preview self.gdrive = gdrive self.ready = True if not os.path.isdir(self.accepted_path): logging.error("Conversion folder not found: %s", self.accepted_path) self.ready = False if not os.path.isdir(self.video_path): logging.error("Video output folder not found: %s", self.video_path) self.ready = False if not os.path.isdir(self.archive_path): logging.error("Processed image archive folder not found: %s", self.archive_path) self.ready = False if not os.path.isdir(self.preview_path): logging.error("Post-processed video preview folder not found: %s", self.preview_path) self.ready = False def convert_file(self, name, image_path, video_path, slate_frame): """Convert a source image file into a video file by animating row by row.""" meta = decode_filename(name) source = read_image_file(image_path) if source is not None: # apply policy limits, possibly adjusting the image source, tempo = validate_animation_timing(source, meta['tempo']) # create a slate sequence generator slate_keyframes = image_keyframes(slate_frame) slate_video_frames = keyframe_interpolator(slate_keyframes) # create an animation sequence generator anim_keyframe_gen = image_row_keyframes(source) anim_video_frames = keyframe_interpolator(anim_keyframe_gen, tempo) # create a composite sequence generator video_frames = itertools.chain(slate_video_frames, anim_video_frames) # open and generate the video file write_video_file(video_path, video_frames) def poll(self): logging.info("ConversionMonitor starting scan...") accepted = os.listdir(self.accepted_path) if len(accepted) > 0: logging.debug("ConversionMonitor found new files: %s", accepted) for name in accepted: # image_path is the full path to the newly accepted image image_path = os.path.join(self.accepted_path, name) # root_name is a tuple (base, extension) root_name = os.path.splitext(name) # the slate identifier is a numeric string used to generate the prefixed color code sequence slate_identifier = calculate_slate_identifer(root_name[0]) # slate_frame is the color code image, code_name is a legible string naming the sequence slate_frame, code_name = color_code_image(slate_identifier) # video_path is the full path to the video file to create for the Pharos system video_path = os.path.join(self.video_path, root_name[0] + '\u2423' + code_name + '.avi') # Convert the video file. logging.info("ConversionMonitor creating %s", video_path) self.convert_file(name, image_path, video_path, slate_frame) # Upload the results to GDrive for the public. if self.gdrive is not None: self.gdrive.upload_approved_image(image_path) # Generate a more-compatible preview version. if os.path.isfile(video_path): # preview_path is the full path to a mp4 version to upload to Google Drive preview_path = os.path.join(self.preview_path, root_name[0] + '\u2423' + code_name + '.mp4') subprocess.run(['ffmpeg', '-y', '-loglevel','quiet', '-i', video_path, '-codec:v', 'libx264', '-pix_fmt', 'yuv420p', '-vf', 'scale=640:32:force_original_aspect_ratio=decrease,pad=640:480:0:224,setsar=1', preview_path]) # Upload the results to GDrive for the public. if self.gdrive is not None and os.path.isfile(preview_path): self.gdrive.upload_preview_video(preview_path) logging.info("ConversionMonitor archiving %s", image_path) subprocess.run(['mv', '-f', image_path, self.archive_path])