Feature/labimp 7672 #42
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,8 +5,8 @@ | |
| from typing import List, Union | ||
|
|
||
| from .. import client_utils, constants, gcs | ||
| from ..client import LabellerrClient | ||
| from ..exceptions import LabellerrError | ||
|
|
||
|
|
||
| def get_total_folder_file_count_and_total_size(folder_path, data_type): | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,8 +21,8 @@ def download(self): | |
| print(f"# Starting batch video processing for dataset: {self.dataset_id}") | ||
| print(f"{'#'*70}\n") | ||
|
|
||
| # Fetch all video files (convert generator to list) | ||
| video_files = list(self.fetch_files()) | ||
|
There was a problem hiding this comment. Converting the generator to a list loads all video files into memory. For large datasets, this could cause memory issues. Consider:
# Process in batches
for video_file in self.fetch_files():
# Process one at a time
video_file.download_create_video_auto_cleanup() |
||
|
|
||
| if not video_files: | ||
| print("No video files found in dataset") | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -35,6 +35,11 @@ def total_frames(self): | |
| """Get total number of frames in the video.""" | ||
| return self.metadata.get("total_frames", 0) | ||
|
|
||
| @property | ||
| def fps(self): | ||
| """Get frames per second of the video.""" | ||
| return self.metadata.get("fps", 25) | ||
|
There was a problem hiding this comment. Potential Bug: Using a hardcoded default FPS of 25 could cause issues. If the actual video has different FPS and this default is used in calculations, it will result in incorrect frame timing. Recommendation: Consider raising an error if FPS is not available in metadata rather than silently falling back to a default, or at minimum log a warning. |
||
|
|
||
| def get_frames(self, frame_start: int = 0, frame_end: int | None = None): | ||
| """ | ||
| Retrieve video frames data from Labellerr API. | ||
|
|
@@ -61,6 +66,7 @@ def get_frames(self, frame_start: int = 0, frame_end: int | None = None): | |
| "frame_end": frame_end, | ||
| "project_id": self.project_id, | ||
| "uuid": unique_id, | ||
| "client_id": self.client.client_id, | ||
| } | ||
|
|
||
| response = self.client.make_request( | ||
|
|
@@ -115,8 +121,15 @@ def download_frames( | |
| :return: Dictionary with download statistics | ||
| """ | ||
| try: | ||
| # Use [Dataset_id]+[File_id]+[File_name] as folder name | ||
|
There was a problem hiding this comment. ✅ Good Improvement: Better Folder Naming Good enhancement to use descriptive folder names with dataset+file+name structure. This makes it much easier to identify videos in the file system. Minor suggestion: Consider documenting this naming convention in the class docstring or module documentation for users. There was a problem hiding this comment. Code Quality Issue: The folder naming logic is overly complex and repeated in multiple places (lines 124-132, 292-298, 315-322, 359-363). Recommendation: Extract this into a private method: def _get_folder_name(self) -> str:
if self.dataset_id and self.file_name:
base_name = os.path.splitext(self.file_name)[0]
return f"{self.dataset_id}+{self.file_id}+{base_name}"
elif self.dataset_id:
return f"{self.dataset_id}+{self.file_id}"
else:
return self.file_idThis reduces code duplication and makes maintenance easier. |
||
| if self.dataset_id and self.file_name: | ||
| # Remove extension from file_name if present | ||
| base_name = os.path.splitext(self.file_name)[0] | ||
| folder_name = f"{self.dataset_id}+{self.file_id}+{base_name}" | ||
| elif self.dataset_id: | ||
| folder_name = f"{self.dataset_id}+{self.file_id}" | ||
| else: | ||
| folder_name = self.file_id | ||
|
|
||
| # Set output path | ||
| if output_folder: | ||
|
|
@@ -207,7 +220,11 @@ def create_video( | |
|
|
||
| input_pattern = os.path.join(frames_folder, pattern) | ||
| if output_file is None: | ||
| # Use [Dataset_id]+[File_id]+[File_name] as default output filename | ||
| if self.dataset_id and self.file_name and self.metadata.get("fps"): | ||
| output_file = f"{self.dataset_id}+{self.file_id}+{self.file_name}+FPS{self.metadata.get('fps')}.mp4" | ||
| else: | ||
| raise ValueError("output_file must be provided") | ||
|
|
||
| # FFmpeg command | ||
| command = [ | ||
|
|
@@ -235,7 +252,7 @@ def create_video( | |
| raise LabellerrError(f"Error while joining frames: {str(e)}") | ||
|
|
||
| def download_create_video_auto_cleanup( | ||
| self, output_folder: str = "./Labellerr_datasets" | ||
|
There was a problem hiding this comment. ✅ Good Fix: Typo Corrected Nice catch fixing the typo from "datastets" to "datasets" in the default parameter! There was a problem hiding this comment. Typo in default path: "Labellerr_datastets" should be "Labellerr_datasets" (missing 'a'). |
||
| ): | ||
| """ | ||
| Download frames, create video, and automatically clean up temporary frames. | ||
|
|
@@ -258,36 +275,52 @@ def download_create_video_auto_cleanup( | |
| print(f"\n[1/4] Fetching frame data from API (0 to {total_frames})...") | ||
| frames_data = self.get_frames(frame_start=0, frame_end=total_frames) | ||
|
|
||
| # print(frames_data) | ||
|
|
||
| if not frames_data: | ||
| raise LabellerrError("No frame data retrieved from API") | ||
|
|
||
| print(f"Retrieved {len(frames_data)} frames") | ||
|
|
||
| # Step 2: Create output folder structure | ||
| print("\n[2/4] Setting up output folders...") | ||
| # Videos will be saved directly in output_folder (labellerr_datasets) | ||
| os.makedirs(output_folder, exist_ok=True) | ||
|
|
||
| # Define actual frames folder path using [Dataset_id]+[File_id]+[File_name] naming | ||
| # Frames will be temporarily stored in a subfolder for organization | ||
| if self.dataset_id and self.file_name: | ||
| base_name = os.path.splitext(self.file_name)[0] | ||
| folder_name = f"{self.dataset_id}+{self.file_id}+{base_name}" | ||
| elif self.dataset_id: | ||
| folder_name = f"{self.dataset_id}+{self.file_id}" | ||
| else: | ||
| folder_name = self.file_id | ||
| actual_frames_folder = os.path.join(output_folder, folder_name) | ||
|
|
||
| # Step 3: Download frames | ||
| print("\n[3/4] Downloading frames...") | ||
| download_result = self.download_frames( | ||
| frames_data=frames_data, output_folder=output_folder | ||
| ) | ||
|
|
||
| if download_result["failed_downloads"] > 0: | ||
| print( | ||
| f"\nWarning: {download_result['failed_downloads']} frames failed to download" | ||
| ) | ||
|
|
||
| # Step 4: Create video from downloaded frames using [Dataset_id]+[File_id]+[File_name]+FPS[fps] naming | ||
| # Save video directly in output_folder (labellerr_datasets) | ||
| print("\n[4/4] Creating video from frames...") | ||
| if self.dataset_id and self.file_name and self.fps: | ||
| # Remove extension from file_name if present, then add FPS and .mp4 | ||
| base_name = os.path.splitext(self.file_name)[0] | ||
| video_filename = ( | ||
| f"{self.dataset_id}+{self.file_id}+{base_name}+FPS{self.fps}.mp4" | ||
| ) | ||
| else: | ||
|
There was a problem hiding this comment. 🐛 Potential Bug: String Formatting Error Line 316 has incorrect string literal - the closing brace should be outside the string: # Current (incorrect):
print("{'='*60}\n") # This prints: {'='*60}
# Should be:
print(f"{'='*60}\n") # This prints: ====================...Or simply: print(f"{'='*60}\n") |
||
| raise ValueError("dataset_id, file_name, and fps metadata are required") | ||
| video_output_path = os.path.join(output_folder, video_filename) | ||
|
|
||
| self.create_video( | ||
| frames_folder=actual_frames_folder, output_file=video_output_path | ||
|
|
@@ -304,7 +337,7 @@ def download_create_video_auto_cleanup( | |
| "file_id": self.file_id, | ||
| "dataset_id": self.dataset_id, | ||
| "video_path": video_output_path, | ||
| "output_folder": output_folder, | ||
| "frames_downloaded": download_result["successful_downloads"], | ||
| "frames_failed": download_result["failed_downloads"], | ||
| "failed_frames_info": download_result["failed_frames"], | ||
|
|
@@ -313,19 +346,22 @@ def download_create_video_auto_cleanup( | |
| print(f"\n{'='*60}") | ||
| print("Processing complete!") | ||
| print(f"Video saved to: {video_output_path}") | ||
| print(f"{'='*60}\n") | ||
|
|
||
| return result | ||
|
|
||
| except Exception as e: | ||
| # Attempt cleanup on error | ||
| # Get the frames folder path using [Dataset_id]+[File_id]+[File_name] naming | ||
| if self.dataset_id is None: | ||
| cleanup_folder = os.path.join(output_folder, self.file_id) | ||
| else: | ||
| if self.file_name: | ||
| base_name = os.path.splitext(self.file_name)[0] | ||
| folder_name = f"{self.dataset_id}+{self.file_id}+{base_name}" | ||
| else: | ||
| folder_name = f"{self.dataset_id}+{self.file_id}" | ||
| cleanup_folder = os.path.join(output_folder, folder_name) | ||
|
|
||
| if os.path.exists(cleanup_folder): | ||
| shutil.rmtree(cleanup_folder) | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Code Quality Issue: The notebook contains hardcoded QA credentials in the config keys (
QA_API_KEY,QA_API_SECRET,QA_CLIENT_ID). While these are loaded from env file, it suggests this notebook is configured for QA environment which shouldn't be in the main branch.Recommendation: Use generic key names or document that this is a development/testing notebook.