partially tested code for the swarm

App/Editor/Schema.py

@@ -9,10 +9,13 @@ class LinkInfo(BaseModel):
 
 
 class Constants(BaseModel):
+    task: Optional[str]
+    chunk: Optional[int]
     duration: Optional[int]
     height: Optional[int]
     width: Optional[int]
     text: Optional[dict]
+    frames: Optional[dict]
 
 
 class Assets(BaseModel):

App/Editor/editorRoutes.py

@@ -1,33 +1,84 @@
-from fastapi import APIRouter, HTTPException, status, BackgroundTasks
+from fastapi import APIRouter, HTTPException, status, BackgroundTasks, UploadFile, Query
 from .Schema import EditorRequest, TaskInfo
-from App.Worker import celery_task
+from App.Worker import celery_task, concatenate_videos
 from celery.result import AsyncResult
+import aiofiles, os, uuid, aiohttp
+from App import SERVER_STATE, Task
 
 videditor_router = APIRouter(tags=["vidEditor"])
 
 
 @videditor_router.post("/create-video")
 async def create_video(videoRequest: EditorRequest, background_task: BackgroundTasks):
-    # background_task.add_task(celery_task, videoRequest)
-    result = celery_task.delay(videoRequest)
+    background_task.add_task(celery_task, videoRequest)
     return {"task_id": "started"}
 
 
-@videditor_router.get("/progress/{task_id}", response_model=TaskInfo)
-async def progress(task_id: str):
-    task_result = AsyncResult(
-        task_id,
-    )
-    if not task_result.ready():
-        progress = task_result.info.get("progress", 0)
-        completed_tasks = task_result.info.get("completed_tasks", [])
+@videditor_router.post("/create-chunks")
+async def create_chunks(videoRequest: EditorRequest, background_task: BackgroundTasks):
+    video_duration = videoRequest.constants.duration
+    task_id = uuid.uuid4()
+    new_task = Task(TASK_ID=task_id)
+
+    active_nodes = [
+        node
+        for node in SERVER_STATE.NODES
+        if await new_task._check_node_online(node.SPACE_HOST)
+    ]
+    number_of_nodes = len(active_nodes)
+    ranges = [
+        [i, i + number_of_nodes] for i in range(0, video_duration, number_of_nodes)
+    ]
+    for i, node in enumerate(active_nodes):
+        await new_task.add_node(node, i)
+
+    SERVER_STATE.TASKS[task_id] = new_task
+
+    async with aiohttp.ClientSession() as session:
+        for i, node in enumerate(active_nodes):
+            videoRequest.constants.frames = ranges[i]
+            if node.SPACE_HOST == SERVER_STATE.SPACE_HOST:
+                background_task.add_task(celery_task, videoRequest)
+            async with session.post(
+                "node.SPACE_HOST/create-video", json=videoRequest
+            ) as response:
+                if response.status != 200:
+                    raise HTTPException(
+                        status_code=response.status,
+                        detail="Failed to post request to node",
+                    )
+
+    return {"task_id": "started"}
+
+
+@videditor_router.post("/uploadfile/")
+async def create_file(
+    background_tasks: BackgroundTasks,
+    file: UploadFile,
+    node: str,
+    chunk: int,
+    task: str,
+):
+
+    chunk_directory = f"/tmp/Video/{task}"
+    file_name = f"{chunk_directory}/{chunk}.mp4"
+    # Create the directory if it does not exist
+    os.makedirs(chunk_directory, exist_ok=True)
+
+    try:
+        async with aiofiles.open(file_name, "wb") as f:
+            while contents := await file.read(1024 * 1):
+                await f.write(contents)
+
+    except Exception as e:
         return {
-            "task_id": task_id,
-            "progress": progress,
-            "completed_tasks": completed_tasks,
+            "message": f"There was an error uploading the file, error message {str(e)}  "
         }
-    else:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail="Task not found",
-        )
+    finally:
+        await file.close()
+    running_task = SERVER_STATE.TASKS[task]
+    running_task.mark_node_completed(node)
+    if running_task.is_completed():
+        background_tasks.add_task(concatenate_videos, chunk_directory)
+
+    return {"message": "File uploaded successfully"}

App/Worker.py

@@ -3,7 +3,7 @@ import os, shutil, subprocess
 import uuid
 from urllib.parse import urlparse
 from subprocess import run
-from App import celery_config, bot
+from App import celery_config, bot, SERVER_STATE
 from typing import List
 from App.Editor.Schema import EditorRequest, LinkInfo, Assets, Constants
 from celery.signals import worker_process_init
@@ -11,6 +11,37 @@ from asgiref.sync import async_to_sync
 import json
 import os
 from pydantic import BaseModel
+from App.utilis import upload_file
+
+import subprocess
+
+
+def concatenate_videos(input_dir):
+    # Get a list of all the mp4 files in the input directory
+    files = sorted([f for f in os.listdir(input_dir) if f.endswith(".mp4")])
+
+    # Generate the input file list for ffmpeg
+    input_files = "|".join([f"file '{os.path.join(input_dir, f)}'" for f in files])
+
+    output_file = f"{input_dir}/final.mp4"
+    # Run the ffmpeg command to concatenate the videos
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-f",
+            "concat",
+            "-safe",
+            "0",
+            "-i",
+            f"concat:{input_files}",
+            "-c",
+            "copy",
+            output_file,
+        ]
+    )
+    bot.start()
+    bot.send_file(-1002069945904, file=output_file, caption="finally done!")
+    return output_file
 
 
 class YouTubeUploadTask(BaseModel):
@@ -23,15 +54,15 @@ class YouTubeUploadTask(BaseModel):
     thumbnail: str = None
 
 
-celery = Celery()
-celery.config_from_object(celery_config)
+# celery = Celery()
+# celery.config_from_object(celery_config)
 # celery.conf.update(
 #     # Other Celery configuration settings
 #     CELERYD_LOG_LEVEL="DEBUG",  # Set log level to DEBUG for the worker
 # )
 
 
-@celery.task(name="CreateFile")
+# @celery.task(name="CreateFile")
 def create_json_file(assets: List[Assets], asset_dir: str):
     for asset in assets:
         filename = f"{asset.type.capitalize()}Sequences.json"
@@ -46,14 +77,22 @@ def create_json_file(assets: List[Assets], asset_dir: str):
             f.write(json_string)
 
 
-@celery.task(name="Constants")
+# @celery.task(name="Constants")
 def create_constants_json_file(constants: Constants, asset_dir: str):
+    temp_dir = asset_dir.replace("src/HelloWorld/Assets", "")
+    instrunction_file = os.path.join(temp_dir, "ServerInstructions.json")
     filename = "Constants.json"
     if constants:
         json_string = json.dumps(constants.dict())
     else:
         json_string = json.dumps({})
     os.makedirs(asset_dir, exist_ok=True)
+    with open(instrunction_file, "w") as f:
+        if constants.instrunctions:
+            f.write(json.dumps({"frames": constants.frames}))
+        else:
+            f.write(json.dumps({"frames": [0, constants.duration]}))
+
     with open(os.path.join(asset_dir, filename), "w") as f:
         f.write(json_string)
 
@@ -73,16 +112,12 @@ def download_with_wget(link, download_dir, filename):
     subprocess.run(["aria2c", link, "-d", download_dir, "-o", filename])
 
 
-@celery.task(name="CopyRemotion")
+# @celery.task(name="CopyRemotion")
 def copy_remotion_app(src: str, dest: str):
     shutil.copytree(src, dest)
 
-    # # create symbolic link to prevent multiple installs
-    # source_dir = os.path.join(src, "node_module")
-    # create_symlink(source_dir, target_dir=dest, symlink_name="node_module")
 
-
-@celery.task(name="Unsilence")
+# @celery.task(name="Unsilence")
 def unsilence(directory: str):
     output_dir = os.path.join(directory, "out/video.mp4")
     shortered_dir = os.path.join(directory, "out/temp.mp4")
@@ -91,13 +126,13 @@ def unsilence(directory: str):
     os.rename(shortered_dir, output_dir)
 
 
-@celery.task(name="InstallDependency")
+# @celery.task(name="InstallDependency")
 def install_dependencies(directory: str):
     os.chdir(directory)
     os.system("npm install")
 
 
-@celery.task(name="uploadTime")
+# @celery.task(name="uploadTime")
 def upload_video_to_youtube(task_data: dict):
     # Convert dict to Pydantic model
     task = YouTubeUploadTask(**task_data)
@@ -128,7 +163,7 @@ def upload_video_to_youtube(task_data: dict):
     return result.stdout
 
 
-@celery.task(name="DownloadAssets")
+# @celery.task(name="DownloadAssets")
 def download_assets(links: List[LinkInfo], temp_dir: str):
     public_dir = f"{temp_dir}/public"
     for link in links:
@@ -137,7 +172,7 @@ def download_assets(links: List[LinkInfo], temp_dir: str):
         download_with_wget(file_link, public_dir, file_name)
 
 
-@celery.task(name="RenderFile")
+# @celery.task(name="RenderFile")
 def render_video(directory: str, output_directory: str):
     os.chdir(directory)
     os.system(
@@ -146,51 +181,75 @@ def render_video(directory: str, output_directory: str):
     print("complete")
 
 
-@celery.task(name="send")
-def cleanup_temp_directory(
-    temp_dir: str, output_dir: str, chat_id: int = -1002069945904
+# @celery.task(name="send")
+async def cleanup_temp_directory(
+    temp_dir: str,
+    output_dir: str,
+    video_task: EditorRequest,
+    chat_id: int = -1002069945904,
 ):
+    video_folder_dir = f"/tmp/Video/{video_task.constants.task}"
     try:
-        print("sending...")
-        bot.start()
-        # bot.send_video(chat_id=chat_id,caption="Your Video Caption",video=output_dir)
-        bot.send_file(chat_id, file=output_dir, caption="Your video caption")
+        if not SERVER_STATE.MASTER:
+            await upload_file(
+                output_dir,
+                SERVER_STATE.SPACE_HOST,
+                video_task.constants.chunk,
+                video_task.constants.task,
+            )
+        else:
+
+            os.makedirs(video_folder_dir, exist_ok=True)
+            shutil.move(
+                output_dir, f"{video_folder_dir}/{video_task.constants.chunk}.mp4"
+            )
+
     except Exception as e:
         print(e)
     finally:
+        remotion_app_dir = os.path.join("/srv", "Remotion-app")
+        shutil.rmtree(remotion_app_dir)
+        # use the cache
+        shutil.copytree(temp_dir, remotion_app_dir)
+        if not SERVER_STATE.CACHED:
+            SERVER_STATE.CACHED = True
         # Cleanup: Remove the temporary directory
         shutil.rmtree(temp_dir, ignore_errors=True)
 
 
-@celery.task(name="All")
-def celery_task(video_task: EditorRequest):
+# @celery.task(name="All")
+async def celery_task(video_task: EditorRequest):
     remotion_app_dir = os.path.join("/srv", "Remotion-app")
     project_id = str(uuid.uuid4())
     temp_dir = f"/tmp/{project_id}"
     output_dir = f"/tmp/{project_id}/out/video.mp4"
     assets_dir = os.path.join(temp_dir, "src/HelloWorld/Assets")
 
-    # copy_remotion_app(remotion_app_dir, temp_dir),
-    # install_dependencies(temp_dir),
-    # create_constants_json_file(video_task.constants, assets_dir),
-    # create_json_file(video_task.assets, assets_dir),
-    # download_assets(video_task.links, temp_dir) if video_task.links else None,
-    # render_video(temp_dir, output_dir),
-    # unsilence(temp_dir),
-    # await cleanup_temp_directory(temp_dir, output_dir),
-
-    chain(
-        copy_remotion_app.si(remotion_app_dir, temp_dir),
-        install_dependencies.si(temp_dir),
-        create_constants_json_file.si(video_task.constants, assets_dir),
-        create_json_file.si(video_task.assets, assets_dir),
-        download_assets.si(video_task.links, temp_dir) if video_task.links else None,
-        render_video.si(temp_dir, output_dir),
-        # unsilence.si(temp_dir),
-        cleanup_temp_directory.si(temp_dir, output_dir),
-    ).apply_async(
-        # link_error=handle_error
-    )  # Link the tasks and handle errors
+    copy_remotion_app(remotion_app_dir, temp_dir)
+
+    # use the cached stuff
+    if not SERVER_STATE.CACHED:
+        install_dependencies(temp_dir)
+
+    create_constants_json_file(video_task.constants, assets_dir)
+    create_json_file(video_task.assets, assets_dir)
+    download_assets(video_task.links, temp_dir)
+    render_video(temp_dir, output_dir)
+    unsilence(temp_dir)
+    await cleanup_temp_directory(temp_dir, output_dir, video_task)
+
+    # chain(
+    #     copy_remotion_app.si(remotion_app_dir, temp_dir),
+    #     install_dependencies.si(temp_dir),
+    #     create_constants_json_file.si(video_task.constants, assets_dir),
+    #     create_json_file.si(video_task.assets, assets_dir),
+    #     download_assets.si(video_task.links, temp_dir) if video_task.links else None,
+    #     render_video.si(temp_dir, output_dir),
+    #     # unsilence.si(temp_dir),
+    #     cleanup_temp_directory.si(temp_dir, output_dir),
+    # ).apply_async(
+    #     # link_error=handle_error
+    # )  # Link the tasks and handle errors
 
 
 def handle_error(task_id, err, *args, **kwargs):

App/__init__.py

@@ -1,12 +1,78 @@
 from pyrogram import Client
 from telethon.sync import TelegramClient
 from telethon.sessions import StringSession
-
+from typing import List, Optional, Dict
 import os
+import aiohttp
+from pydantic import BaseModel
+
+
+class Node(BaseModel):
+    MASTER: bool
+    SPACE_HOST: str
+
+
+class NodeTaskState(BaseModel):
+    NODE: Node
+    CHUNK: int
+    COMPLETED: bool
+
+
+class Task(BaseModel):
+    TASK_ID: str
+    COMPLETED: bool = False
+    NODES: Optional[List[NodeTaskState]] = []
+
+    def is_completed(self) -> bool:
+        for node in self.NODES:
+            if not node.COMPLETED:
+                return False
+        self.COMPLETED = True
+        return True
+
+    def mark_node_completed(self, space_host: str):
+        for state in self.NODES:
+            if state.NODE.SPACE_HOST == space_host:
+                state.COMPLETED = True
+                break
 
+    async def add_node(self, node: Node, CHUNK: int):
+        new_node_state = NodeTaskState(NODE=node, CHUNK=CHUNK, COMPLETED=False)
+        self.NODES.append(new_node_state)
 
-TELEGRAM_SESSION = os.environ.get("TELEGRAM_SESSION")
+    @classmethod
+    async def _check_node_online(cls, space_host: str) -> bool:
+        try:
+            async with aiohttp.ClientSession(
+                timeout=aiohttp.ClientTimeout(total=5)
+            ) as session:
+                async with session.get(space_host) as response:
+                    return response.status == 200
+        except aiohttp.ClientError:
+            return False
+
+
+class ServerState(Node):
+    CACHED: bool
+    TASKS: Optional[Dict[str, Task]] = {}
+    NODES: Optional[list[Node]]
+    DB: str = "https://herokuserver-185316.firebaseio.com/"
+
+    def get_master(self) -> Optional[Node]:
+        for node in self.NODES:
+            if node.NODE.MASTER:
+                return node.NODE
+        return None
+
+
+TELEGRAM_SESSION = os.environ.get("TELEGRAM_SESSION", "RANDOM_STRING")
 TELEGRAM_SESSION_PYROGRAM = os.environ.get("TELEGRAM_SESSION_PYROGRAM")
+MASTER_SERVER = bool(os.environ.get("MASTER", 0))
+SPACE_HOST = os.environ.get("SPACE_HOST", "RANDOM_STRING")
+
+
+SERVER_STATE = ServerState(CACHED=False, MASTER=MASTER_SERVER, SPACE_HOST=SPACE_HOST)
+
 
 bot: TelegramClient = TelegramClient(
     StringSession(TELEGRAM_SESSION),

App/app.py

@@ -1,18 +1,19 @@
 from fastapi import FastAPI, BackgroundTasks
 from .Editor.editorRoutes import videditor_router
 from App import bot
+from App.utilis import WorkerClient, SERVER_STATE
 
 app = FastAPI()
+manager = WorkerClient()
 
 
-# @app.on_event("startup")
-# async def startup_event():
-#     await bot.start()
-
-
-# @app.on_event("shutdown")
-# async def shutdown_event():
-#     await bot.stop()
+@app.on_event("startup")
+async def startup_event():
+    if SERVER_STATE.MASTER:
+        await bot.start()
+    response = await manager.register_worker()
+    if not response:
+        print("Error registering worker")
 
 
 @app.get("/")

App/utilis.py

@@ -0,0 +1,60 @@
+import aiohttp, asyncio
+from App import SERVER_STATE, Node
+
+import aiohttp
+
+
+async def upload_file(file_path: str, node: str, chunk: int, task: str):
+    master_node = SERVER_STATE.get_master()
+    url = f"http://{master_node.SPACE_HOST}/uploadfile/?node={node}&chunk={chunk}&task={task}"
+    async with aiohttp.ClientSession() as session:
+        headers = {"Transfer-Encoding": "chunked"}
+        with open(file_path, "rb") as file:
+            async with session.post(url, headers=headers, data=file) as response:
+                if response.status == 200:
+                    print("File uploaded successfully")
+                else:
+                    print("Failed to upload file")
+
+
+class WorkerClient:
+    base_url = SERVER_STATE.DB
+
+    async def register_worker(self):
+        async with aiohttp.ClientSession() as session:
+            data = {
+                "WORKER_ID": SERVER_STATE.SPACE_HOST,
+                "MASTER": SERVER_STATE.MASTER,
+                "HOST_NAME": SERVER_STATE.SPACE_HOST,
+                "SPACE_HOST": SERVER_STATE.SPACE_HOST,
+            }
+            response = await self.get_node()
+            if response:
+                return response
+
+            async with session.put(
+                f"{self.base_url}/nodes/{SERVER_STATE.SPACE_HOST}.json", json=data
+            ) as resp:
+                return await resp.json()
+
+    async def get_node(self):
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                f"{self.base_url}/nodes/{SERVER_STATE.SPACE_HOST}.json"
+            ) as resp:
+                response = await resp.json()
+                return response
+
+    async def delete_node(self):
+        async with aiohttp.ClientSession() as session:
+            async with session.delete(
+                f"{self.base_url}/nodes/{SERVER_STATE.SPACE_HOST}.json"
+            ) as resp:
+                response = await resp.json()
+
+    async def get_all_nodes(self):
+        async with aiohttp.ClientSession() as session:
+            async with session.get(f"{self.base_url}/nodes.json") as resp:
+                response = await resp.json()
+                SERVER_STATE.NODES = [Node(**value) for value in response.values()]
+                return SERVER_STATE.NODES

Dockerfile

@@ -1,17 +1,22 @@
 # Builder stage
 FROM python:3.10.0 as builder
 
+# Create a non-root user
 RUN useradd -ms /bin/bash admin
 
+# Set the working directory
 WORKDIR /srv
-RUN chown -R admin:admin /srv
-RUN chmod -R 755 /srv
 
-# Install dependencies
-RUN apt-get update && \
-  apt-get install -y   wget ffmpeg curl aria2
+# Copy requirements file first to leverage caching
+COPY --chown=admin requirements.txt .
 
-RUN apt-get install -y \
+# Install Python dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Install system dependencies
+RUN apt-get update && \
+  apt-get install -y --no-install-recommends \
+  wget ffmpeg curl aria2 \
   fonts-liberation \
   libatk-bridge2.0-0 \
   libatk1.0-0 \
@@ -26,63 +31,38 @@ RUN apt-get install -y \
   libvulkan1 \
   libxcomposite1 \
   libxdamage1 \
-  mesa-vulkan-drivers\
+  mesa-vulkan-drivers \
   libxfixes3 \
   libasound2 \
   libxkbcommon0 \
   libxrandr2 \
-  xdg-utils
+  xdg-utils \
+  software-properties-common \
+  npm
 
-# Download youtubeuploader
+# Install youtubeuploader
 ADD https://github.com/porjo/youtubeuploader/releases/download/23.06/youtubeuploader_23.06_Linux_x86_64.tar.gz youtubeuploader.tar.gz
-
-
-# Create youtube directory and extract youtubeuploader there
 RUN mkdir -p /srv/youtube && \
   tar -zxvf youtubeuploader.tar.gz -C /srv/youtube && \
   rm youtubeuploader.tar.gz && \
   chmod +x /srv/youtube/youtubeuploader
 
-
-# Copy the application code
-COPY --chown=admin . /srv
-
-# Download and install Thorium Browser
-
-RUN apt-get update && apt-get install -y \
-  software-properties-common \
-  npm
+# Install latest npm and node
 RUN npm install npm@latest -g && \
   npm install n -g && \
   n latest
 
+# Install unsilence
+RUN pipx ensurepath && \
+  pipx install unsilence
 
-
-
-#install the stuff
-# RUN cd /srv/Remotion-app && npm install
-
-
-
-
-
-# Install Python dependencies
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
-
-#install unsilence
-RUN pipx ensurepath
-RUN pipx install unsilence
-
-
-
-
+# Copy the application code
+COPY --chown=admin . /srv
 
 # Command to run the application
-CMD python -m uvicorn App.app:app --host 0.0.0.0 --port 7860 &  python -m celery -A App.Worker.celery worker -c 5  --max-tasks-per-child=1  --without-heartbeat 
-
-
-# CMD python -m uvicorn App.app:app --host 0.0.0.0  --port 7860  --workers 2
+# CMD python -m uvicorn App.app:app --host 0.0.0.0 --port 7860 &  python -m celery -A App.Worker.celery worker -c 5  --max-tasks-per-child=1  --without-heartbeat 
 
+CMD python -m uvicorn App.app:app --host 0.0.0.0 --port 7860 
 
+# Expose port
 EXPOSE 7860

Remotion-app/ServerInstructions.json

@@ -0,0 +1,3 @@
+{
+	"frames": [0, 30]
+}

Remotion-app/remotion.config.js

@@ -4,11 +4,13 @@
 // Note: When using the Node.JS APIs, the config file doesn't apply. Instead, pass options directly to the APIs
 
 import {Config} from '@remotion/cli/config';
-
+import Instructions from './ServerInstructions.json';
 import {enableTailwind} from '@remotion/tailwind';
 Config.overrideWebpackConfig((currentConfiguration) => {
 	return enableTailwind(currentConfiguration);
 });
 
+Config.setC;
 Config.setVideoImageFormat('jpeg');
+Config.setFrameRange(Instructions.frames);
 // Config.setConcurrency(15);

requirements.txt

@@ -9,6 +9,7 @@ starlette==0.25.0
 Werkzeug==2.2.2
 uvicorn==0.21.1
 gunicorn
+aiofiles
 requests
 celery==5.3.0
 pyrogram==2.0.100