Multicam

Use the python script below to process videos from two cameras. Place your API key and the location of the video files.

end_to_end_2cams_processing.py

import subprocess, sys

#Install 
subprocess.check_call([sys.executable, "-m", "pip", "install", "move-ugc-python"])

from move_ugc import MoveUgc
from move_ugc.schemas.sources import SourceIn
from move_ugc.schemas.sync_method import SyncMethodInput, ClapWindowInput
from move_ugc.schemas.volume import AreaType
from datetime import datetime
import os
from pathlib import Path
import time
import requests

class MoveAI:
    """MoveAI UGC utility class."""

    def __init__(self, api_key, endpoint_url=None) -> None:
        """Initialize MoveAI UGC utility class.

        Args:
            api_key: API key.
            endpoint_url: Endpoint
        """
        self.api_key = api_key
        if endpoint_url is None:
            endpoint_url = 'https://api.move.ai/ugc/graphql'
        self.endpoint_url = endpoint_url
        self.ugc_client = MoveUgc(api_key=api_key, endpoint_url=endpoint_url)

    def get_client(self):
        """Get MoveUGC ugc_client.

        Returns:
            MoveUgc
        """
        client = self.ugc_client.client.retrieve()
        return client

    def create_files(self, video_path: str) -> str:
        """Create a file in MoveUGC.

        Args:
            video_path: Path to the video file.

        Returns:
            str: File ID.
        """
        video_file = self.ugc_client.files.create(file_type="mp4")
        print("File created:", video_file.id)

        with open(video_path, 'rb') as f:
            print("Uploading...")
            requests.put(video_file.presigned_url, data=f.read())

        return video_file.id

    def create_volume(
        self, sources, human_height: float, name=None, metadata=None,
    ):
        """Create a new volume.

        Args:
            sources: List of sources.
            human_height: Human height.
            name: Name of the volume.
            metadata: Metadata.

        Returns:
            VolumeType
        """
        clap_window = ClapWindowInput(start_time=0.1, end_time=5.0)
        sync_method = SyncMethodInput(clap_window=clap_window)
        if metadata is None:
            metadata = {"test": "Multicam Quickstart"}
        return self.ugc_client.volumes.create_human_volume(
            sources=sources,
            name=name,
            metadata=metadata,
            sync_method=sync_method,
            human_height=human_height,
            area_type=AreaType.NORMAL,
        )

    def get_volume(self, volume_id: str):
        """Retrieve volume.

        Args:
            VolumeType
        """
        return self.ugc_client.volumes.retrieve_human_volume(id=volume_id)

    def create_take(self, sources, volume_id: str, sync_method, name=None, metadata=None) -> str:
        """Create a new take.

        Args:
            sources: List of sources.
            volume_id: Volume ID.
            name: Name of the take.
            metadata: Metadata.

        Returns:
            str: Take ID.
        """
        if metadata is None:
            metadata = {"test": "Multicam Quickstart"}
        return self.ugc_client.takes.create_multicam(
            volume_id=volume_id,
            sources=sources,
            metadata=metadata,
            name=name,
            sync_method=sync_method,
        ).id

    def create_job(
        self, take_id: str, number_of_actors: str, name=None, metadata=None,
    ):
        """Create a new multicam job.

        Args:
            take_id: Take ID.
            number_of_actors: Number of actors.
            name: Name of the job.
            metadata: Metadata.

        Returns:
            JobType.
        """
        return self.ugc_client.jobs.create_multicam(
            take_id=take_id, number_of_actors=number_of_actors, metadata={"test": "Multicam Quickstart"},
        )

    def get_job(self, job_id, expand=False):
        # Get a job using the Move One Public API
        # Implement job retrieval logic using move_ugc_python SDK
        if expand is False:
            job = self.ugc_client.jobs.retrieve(id=job_id)
        else:
            job = self.ugc_client.jobs.retrieve(
                id=job_id, expand=["take", "outputs"]
            )
        return job

    def download_outputs(self, job_id, output_dir, output_name):
        # make output dir if it doesn't exist
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
        # get job
        job = self.get_job(job_id, expand=True)
        # for each output download the file in the output_dir
        output_paths = []
        for output in job.outputs:
            output_file_name = f"{output_name}{output.file.type}"
            output_path = os.path.join(output_dir, output_file_name)
            with open(output_path, 'wb') as f:
                response = requests.get(output.file.presigned_url)
                f.write(response.content)
            output_paths.append(output_path)
        return output_paths
    

ugc_client = MoveAI("<YOUR API KEY")

if not ugc_client.api_key :
    raise ValueError('Please set the MOVE_API_KEY as an argument to MoveAI')

sources = []
# Create 2 files to upload calibration videos
for camera_number in range(1, 3):
    # Creates and uploads cam01_calib.mp4, cam0N_calib.mp4 etc.
    input_video_file = Path(f'data/input_videos/calib/cam0{camera_number}_calib.mp4')
    ugc_file = ugc_client.create_files(input_video_file)
    print("File uploaded:", ugc_file, "appending to sources...")
    sources.append(
        SourceIn(
            device_label=f"cam0{camera_number}",
            file_id=ugc_file,
            format="MP4",
            camera_settings={
                "lens": "goprohero10-fhd", 
            }
        )
    )

volume = ugc_client.create_volume(sources=sources, human_height=1.77, name="Multicam Quickstart")
print("Volume created:", volume.id)
# Poll the volume until it is finished processing
attempts = 0
print("Polling volume...")
while attempts < 300:
    volume = ugc_client.get_volume(volume.id)
    update_str = f"[{datetime.now().isoformat()} | {attempts}] Volume {volume.id} is {volume.state}"
    print(update_str)
    if volume.state == 'FINISHED':
        print("Volume is processed successfully, please proceed to job creation")
        break
    else:
        time.sleep(30)
        attempts += 1


#Create files for action take
sources = []
for camera_number in range(1,3):
    # Creates and uploads cam01_action.mp4, cam0N_action.mp4 etc.
    input_video_file = Path(f'data/input_videos/action/cam0{camera_number}_action.mp4')
    ugc_file = ugc_client.create_files(input_video_file)
    print("File uploaded:", ugc_file, "appending to sources...")
    sources.append(
        SourceIn(
            device_label=f"cam0{camera_number}",
            file_id=ugc_file,
            format="MP4",
            camera_settings={
                "lens": "goprohero8-fhd",
            }
        )
    )


take_id = ugc_client.create_take(
    sources, 
    volume_id=volume.id, 
    sync_method=SyncMethodInput(
        clap_window={
            "start_time": 0.1,
            "end_time": 3.0,
        },
    ),
    name="Test take"
)

print("Take created:", take_id)


job = ugc_client.create_job(take_id=take_id, number_of_actors=1, name="Quickstart job")
print("Job Created:", job.id)

# Poll the job until it is finished
attempts = 0
output_dir = Path('data/output')
while attempts < 300:
    job = ugc_client.get_job(job.id)
    update_str = f"[{datetime.now().isoformat()} | {attempts}] Job {job.id} is {job.state}"
    if job.state == 'FINISHED':
        outputs = ugc_client.download_outputs(job.id, output_dir, "Multicam_demo.mp4")
        print(f"Outputs downloaded to {output_dir}")
        print(f"Output files: {outputs}")
        break
    else:
        time.sleep(60)
        attempts += 1