Process videos and images to compute condition on videos. Args: input_path: Path to input video/image file or directory output_path: Path to output reference video file override: Whether to override existing reference video files
(
input_path: Path,
output_path: Path,
override: bool,
batch_size: int = 100,
)
| 140 | |
| 141 | |
| 142 | def process_media( |
| 143 | input_path: Path, |
| 144 | output_path: Path, |
| 145 | override: bool, |
| 146 | batch_size: int = 100, |
| 147 | ) -> None: |
| 148 | """Process videos and images to compute condition on videos. |
| 149 | Args: |
| 150 | input_path: Path to input video/image file or directory |
| 151 | output_path: Path to output reference video file |
| 152 | override: Whether to override existing reference video files |
| 153 | """ |
| 154 | if not output_path.exists(): |
| 155 | raise FileNotFoundError( |
| 156 | f"Output file does not exist: {output_path}. This is also the input file for the dataset." |
| 157 | ) |
| 158 | |
| 159 | # Check for existing reference video files |
| 160 | meta_data = _get_meta_data(output_path) |
| 161 | |
| 162 | base_dir = input_path.resolve() |
| 163 | console.print(f"Using [bold blue]{base_dir}[/] as base directory for relative paths") |
| 164 | |
| 165 | # Filter media files |
| 166 | media_to_process = [] |
| 167 | skipped_media = [] |
| 168 | |
| 169 | def media_path_to_reference_path(media_file: Path) -> Path: |
| 170 | return media_file.parent / (media_file.stem + "_reference" + media_file.suffix) |
| 171 | |
| 172 | media_files = [base_dir / Path(_get_media_path(sample)) for sample in meta_data] |
| 173 | for media_file in media_files: |
| 174 | reference_path = media_path_to_reference_path(media_file) |
| 175 | media_to_process.append(media_file) |
| 176 | |
| 177 | console.print(f"Processing [bold]{len(media_to_process)}[/] media.") |
| 178 | |
| 179 | # Initialize progress tracking |
| 180 | progress = Progress( |
| 181 | SpinnerColumn(), |
| 182 | TextColumn("{task.description}"), |
| 183 | BarColumn(bar_width=40), |
| 184 | MofNCompleteColumn(), |
| 185 | TimeElapsedColumn(), |
| 186 | TextColumn("•"), |
| 187 | TimeRemainingColumn(), |
| 188 | console=console, |
| 189 | ) |
| 190 | |
| 191 | # Process media files |
| 192 | media_paths = [_get_media_path(item) for item in meta_data] |
| 193 | reference_paths = {rel_path: str(media_path_to_reference_path(Path(rel_path))) for rel_path in media_paths} |
| 194 | |
| 195 | with progress: |
| 196 | task = progress.add_task("Computing condition on videos", total=len(media_to_process)) |
| 197 | |
| 198 | for media_file, rel_path in zip(media_to_process, media_paths, strict=True): |
| 199 | progress.update(task, description=f"Processing [bold blue]{media_file.name}[/]") |
no test coverage detected