import json from typing import Any, Dict, List, Union from enum import Enum import collections.abc from shortGPT.editing_framework.core_editing_engine import CoreEditingEngine def update_dict(d, u): for k, v in u.items(): if isinstance(v, collections.abc.Mapping): d[k] = update_dict(d.get(k, {}), v) else: d[k] = v return d class EditingStep(Enum): CROP_1920x1080 = "crop_1920x1080_to_short.json" ADD_CAPTION_SHORT = "make_caption.json" ADD_CAPTION_SHORT_ARABIC = "make_caption_arabic.json" ADD_CAPTION_LANDSCAPE = "make_caption_landscape.json" ADD_CAPTION_LANDSCAPE_ARABIC = "make_caption_arabic_landscape.json" ADD_WATERMARK = "show_watermark.json" ADD_SUBSCRIBE_ANIMATION = "subscribe_animation.json" SHOW_IMAGE = "show_top_image.json" ADD_VOICEOVER_AUDIO = "add_voiceover.json" ADD_BACKGROUND_MUSIC = "background_music.json" ADD_REDDIT_IMAGE = "show_reddit_image.json" ADD_BACKGROUND_VIDEO = "add_background_video.json" INSERT_AUDIO = "insert_audio.json" EXTRACT_AUDIO = "extract_audio.json" ADD_BACKGROUND_VOICEOVER = "add_background_voiceover.json" class Flow(Enum): WHITE_REDDIT_IMAGE_FLOW = "build_reddit_image.json" from pathlib import Path _here = Path(__file__).parent STEPS_PATH = (_here / 'editing_steps/').resolve() FLOWS_PATH = (_here / 'flows/').resolve() class EditingEngine: def __init__(self,): self.editing_step_tracker = dict((step, 0) for step in EditingStep) self.schema = {'visual_assets': {}, 'audio_assets': {}} def addEditingStep(self, editingStep: EditingStep, args: Dict[str, any] = {}): json_step = json.loads( open(STEPS_PATH / f"{editingStep.value}", 'r', encoding='utf-8').read()) step_name, editingStepDict = list(json_step.items())[0] if 'inputs' in editingStepDict: required_args = (editingStepDict['inputs']['actions'] if 'actions' in editingStepDict['inputs'] else []) + (editingStepDict['inputs']['parameters'] if 'parameters' in editingStepDict['inputs'] else []) for required_argument in required_args: if required_argument not in args: raise Exception( f"Error. '{required_argument}' input missing, you must include it to use this editing step") if required_args: pass action_names = [action['type'] for action in editingStepDict['actions'] ] if 'actions' in editingStepDict else [] param_names = [param_name for param_name in editingStepDict['parameters'] ] if 'parameters' in editingStepDict else [] for arg_name in args: if ('inputs' in editingStepDict): if 'parameters' in editingStepDict['inputs'] and arg_name in param_names: editingStepDict['parameters'][arg_name] = args[arg_name] pass if 'actions' in editingStepDict['inputs'] and arg_name in action_names: for i, action in enumerate(editingStepDict['actions']): if action['type'] == arg_name: editingStepDict['actions'][i]['param'] = args[arg_name] if editingStepDict['type'] == 'audio': self.schema['audio_assets'][f"{step_name}_{self.editing_step_tracker[editingStep]}"] = editingStepDict else: self.schema['visual_assets'][f"{step_name}_{self.editing_step_tracker[editingStep]}"] = editingStepDict self.editing_step_tracker[editingStep] += 1 def ingestFlow(self, flow: Flow, args): json_flow = json.loads(open(FLOWS_PATH / f"{flow.value}", 'r', encoding='utf-8').read()) for required_argument in list(json_flow['inputs'].keys()): if required_argument not in args: raise Exception( f"Error. '{required_argument}' input missing, you must include it to use this editing step") update = args[required_argument] for path_key in reversed(json_flow['inputs'][required_argument].split("/")): update = {path_key: update} json_flow = update_dict(json_flow, update) self.schema = json_flow def dumpEditingSchema(self): return self.schema def renderVideo(self, outputPath, logger=None): engine = CoreEditingEngine() engine.generate_video(self.schema, outputPath, logger=logger) def renderImage(self, outputPath, logger=None): engine = CoreEditingEngine() engine.generate_image(self.schema, outputPath, logger=logger) def generateAudio(self, outputPath, logger=None): engine = CoreEditingEngine() engine.generate_audio(self.schema, outputPath, logger=logger) # import json # from typing import Any, Dict, List, Union # from enum import Enum # import collections.abc # import os # from shortGPT.editing_framework.core_editing_engine import CoreEditingEngine # def update_dict(d, u): # for k, v in u.items(): # if isinstance(v, collections.abc.Mapping): # d[k] = update_dict(d.get(k, {}), v) # else: # d[k] = v # return d # class EditingStep(Enum): # CROP_1920x1080 = "crop_1920x1080_to_short.json" # ADD_CAPTION_SHORT = "make_caption.json" # ADD_CAPTION_SHORT_ARABIC = "make_caption_arabic.json" # ADD_CAPTION_LANDSCAPE = "make_caption_landscape.json" # ADD_CAPTION_LANDSCAPE_ARABIC = "make_caption_arabic_landscape.json" # ADD_WATERMARK = "show_watermark.json" # ADD_SUBSCRIBE_ANIMATION = "subscribe_animation.json" # SHOW_IMAGE = "show_top_image.json" # ADD_VOICEOVER_AUDIO = "add_voiceover.json" # ADD_BACKGROUND_MUSIC = "background_music.json" # ADD_REDDIT_IMAGE = "show_reddit_image.json" # ADD_BACKGROUND_VIDEO = "add_background_video.json" # INSERT_AUDIO = "insert_audio.json" # EXTRACT_AUDIO = "extract_audio.json" # ADD_BACKGROUND_VOICEOVER = "add_background_voiceover.json" # class Flow(Enum): # WHITE_REDDIT_IMAGE_FLOW = "build_reddit_image.json" # STEPS_PATH = "shortGPT/editing_framework/editing_steps/" # FLOWS_PATH = "shortGPT/editing_framework/flows/" # class EditingTrack: # def __init__(self, filepath=None): # self.editing_step_tracker = dict((step, 0) for step in EditingStep) # self.schema = {'visual_assets': {}, 'audio_assets': {}} # self.filepath = filepath # if filepath is not None: # try: # self.load_from_file(filepath) # except FileNotFoundError: # self.save_to_file(filepath) # def addEditingStep(self, editingStep: EditingStep, args: Dict[str, any] = {}): # json_step = json.loads( # open(STEPS_PATH+editingStep.value, 'r', encoding='utf-8').read()) # step_name, editingStepDict = list(json_step.items())[0] # if 'inputs' in editingStepDict: # required_args = (editingStepDict['inputs']['actions'] if 'actions' in editingStepDict['inputs'] else []) + (editingStepDict['inputs']['parameters'] if 'parameters' in editingStepDict['inputs'] else []) # for required_argument in required_args: # if required_argument not in args: # raise Exception( # f"Error. '{required_argument}' input missing, you must include it to use this editing step") # if required_args: # pass # action_names = [action['type'] for action in editingStepDict['actions'] # ] if 'actions' in editingStepDict else [] # param_names = [param_name for param_name in editingStepDict['parameters'] # ] if 'parameters' in editingStepDict else [] # for arg_name in args: # if ('inputs' in editingStepDict): # if 'parameters' in editingStepDict['inputs'] and arg_name in param_names: # editingStepDict['parameters'][arg_name] = args[arg_name] # pass # if 'actions' in editingStepDict['inputs'] and arg_name in action_names: # for i, action in enumerate(editingStepDict['actions']): # if action['type'] == arg_name: # editingStepDict['actions'][i]['param'] = args[arg_name] # if editingStepDict['type'] == 'audio': # self.schema['audio_assets'][f"{step_name}_{self.editing_step_tracker[editingStep]}"] = editingStepDict # else: # self.schema['visual_assets'][f"{step_name}_{self.editing_step_tracker[editingStep]}"] = editingStepDict # self.editing_step_tracker[editingStep] += 1 # def ingestFlow(self, flow: Flow, args): # json_flow = json.loads(open(FLOWS_PATH+flow.value, 'r', encoding='utf-8').read()) # for required_argument in list(json_flow['inputs'].keys()): # if required_argument not in args: # raise Exception( # f"Error. '{required_argument}' input missing, you must include it to use this editing step") # update = args[required_argument] # for path_key in reversed(json_flow['inputs'][required_argument].split("/")): # update = {path_key: update} # json_flow = update_dict(json_flow, update) # self.schema = json_flow # def dumpEditingSchema(self): # return self.schema # def save_to_file(self): # if self.file_path: # with open(self.file_path, 'w') as f: # json.dump({'step_tracker': {key.name: value for key, value in self.step_tracker.items()}, 'asset_schema': self.asset_schema}, f) # def load_from_file(self): # if self.file_path and os.path.exists(self.file_path): # with open(self.file_path, 'r') as f: # data = json.load(f) # self.step_tracker = {EditingStep[key]: value for key, value in data.get('step_tracker', {}).items()} # self.asset_schema = data.get('asset_schema', {'visual_assets': {}, 'audio_assets': {}}) # else: # raise Exception("File does not exist") # def renderVideo(self, outputPath, logger=None): # engine = CoreEditingEngine() # engine.generate_video(self.schema, outputPath, logger=logger) # def renderImage(self, outputPath, logger=None): # engine = CoreEditingEngine() # engine.generate_image(self.schema, outputPath, logger=logger) # def generateAudio(self, outputPath, logger=None): # engine = CoreEditingEngine() # engine.generate_audio(self.schema, outputPath, logger=logger)