PTDT-3807: Add temporal audio annotation support by rishisurana-labelbox · Pull Request #2013 · Labelbox/labelbox-python
#!/usr/bin/env python3 import labelbox as lb import uuid from labelbox.data.serialization.ndjson.label import NDLabel import labelbox.types as lb_types from labelbox.data.annotation_types.label import Label import argparse # Parse command line arguments parser = argparse.ArgumentParser(description='Compare and optionally upload temporal audio classifications') parser.add_argument('--upload-raw', action='store_true', help='Upload raw predictions from actual.py') parser.add_argument('--upload-computed', action='store_true', help='Upload computed predictions from class-based API') parser.add_argument('--env', type=str, choices=['local', 'prod'], default='local', help='Environment to upload to (local or prod)') args = parser.parse_args() asset = { "row_data": "https://storage.googleapis.com/lb-artifacts-testing-public/audio/gpt_how_can_you_help_me.wav", "global_key": "123", "media_type": "AUDIO", } # Step 2: Create ontology with temporal classifications (SAME AS NDJSON VERSION) ontology_builder = lb.OntologyBuilder( tools=[ # No tools needed for audio temporal - classifications handle everything ], classifications=[ # Text with nested classifications lb.Classification( class_type=lb.Classification.Type.TEXT, name="text_class", scope=lb.Classification.Scope.INDEX, options=[ lb.Classification( class_type=lb.Classification.Type.TEXT, name="text_text", options=[ lb.Classification( class_type=lb.Classification.Type.RADIO, name="text_text_radio", options=[ lb.Option("text_text_radio_option_1"), lb.Option("text_text_radio_option_2"), ], ), lb.Classification( class_type=lb.Classification.Type.TEXT, name="text_text_text", ), ] ), lb.Classification( class_type=lb.Classification.Type.RADIO, name="text_radio", options=[ lb.Option( "text_radio_option_1", options=[ lb.Classification( class_type=lb.Classification.Type.CHECKLIST, name="text_radio_checklist", options=[ lb.Option("text_radio_checklist_option_1"), lb.Option("text_radio_checklist_option_2"), lb.Option("text_radio_checklist_option_3"), ] ) ] ), lb.Option("text_radio_option_2"), ] ), lb.Classification( class_type=lb.Classification.Type.CHECKLIST, name="text_checklist", options=[ lb.Option( "text_checklist_option_1", options=[ lb.Classification( class_type=lb.Classification.Type.TEXT, name="text_checklist_text", ) ] ), lb.Option("text_checklist_option_2"), lb.Option("text_checklist_option_3"), ] ) ], ), # Radio with nested classifications lb.Classification( class_type=lb.Classification.Type.RADIO, name="radio_class", scope=lb.Classification.Scope.INDEX, options=[ lb.Option( "radio_class_option_1", options=[ lb.Classification( class_type=lb.Classification.Type.TEXT, name="radio_text", options=[ lb.Classification( class_type=lb.Classification.Type.CHECKLIST, name="radio_text_checklist", options=[ lb.Option("option_1_radio_text_checklist"), lb.Option("option_2_radio_text_checklist"), lb.Option("option_3_radio_text_checklist"), ], ) ] ) ], ), lb.Option( "radio_class_option_2", options=[ lb.Classification( class_type=lb.Classification.Type.RADIO, name="radio_radio", options=[ lb.Option( "radio_radio_option_1", options=[ lb.Classification( class_type=lb.Classification.Type.TEXT, name="radio_radio_text", ) ] ), lb.Option("radio_radio_option_2"), ], ) ] ), lb.Option( "radio_class_option_3", options=[ lb.Classification( class_type=lb.Classification.Type.CHECKLIST, name="radio_checklist", options=[ lb.Option( "radio_checklist_option_1", options=[ lb.Classification( class_type=lb.Classification.Type.RADIO, name="radio_checklist_radio", options=[ lb.Option("radio_checklist_radio_option_1"), lb.Option("radio_checklist_radio_option_2"), ], ) ] ), lb.Option("radio_checklist_option_2"), ], ) ] ), ], ), # Checklist with nested classifications lb.Classification( class_type=lb.Classification.Type.CHECKLIST, name="checklist_class", scope=lb.Classification.Scope.INDEX, options=[ lb.Option( "checklist_class_option_1", options=[ lb.Classification( class_type=lb.Classification.Type.TEXT, name="checklist_text", options=[ lb.Classification( class_type=lb.Classification.Type.TEXT, name="checklist_text_text" ) ] ) ], ), lb.Option( "checklist_class_option_2", options=[ lb.Classification( class_type=lb.Classification.Type.RADIO, name="checklist_radio", options=[ lb.Option( "option_1_checklist_radio", options=[ lb.Classification( class_type=lb.Classification.Type.RADIO, name="checklist_radio_radio", options=[ lb.Option("option_1_checklist_radio_radio"), lb.Option("option_2_checklist_radio_radio"), ], ) ] ), lb.Option("option_2_checklist_radio"), ], ) ], ), lb.Option( "checklist_class_option_3", options=[ lb.Classification( class_type=lb.Classification.Type.CHECKLIST, name="checklist_checklist", options=[ lb.Option( "option_1_checklist_checklist", options=[ lb.Classification( class_type=lb.Classification.Type.CHECKLIST, name="checklist_checklist_checklist", options=[ lb.Option("option_1_checklist_checklist_checklist"), lb.Option("option_2_checklist_checklist_checklist"), lb.Option("option_3_checklist_checklist_checklist"), ] ) ] ), lb.Option("option_2_checklist_checklist"), lb.Option("option_3_checklist_checklist"), ], ) ], ), ], ), ], ) # Step 5: Create temporal classifications using class-based API # TEXT_CLASS: Text > Text/Radio/Checklist branches text_class_annotation = lb_types.TemporalClassificationText( name="text_class", value=[ # text_text branch with nested radio (100, 1200, "top text with text classification"), # text_checklist branch with nested checklist (1400, 2400, "top level text with checklist classification"), # text_radio branch with nested text (2600, 3500, "text with radio"), ], classifications=[ # text_text nested classification lb_types.TemporalClassificationText( name="text_text", value=[ (200, 1000, "nested text classification with nested radio"), ], classifications=[ # text_text_radio nested radio lb_types.TemporalClassificationQuestion( name="text_text_radio", value=[ lb_types.TemporalClassificationAnswer( name="text_text_radio_option_1", frames=[(300, 500)], ), lb_types.TemporalClassificationAnswer( name="text_text_radio_option_2", frames=[(501, 900)], ), ], ), # text_text_text nested text lb_types.TemporalClassificationText( name="text_text_text", value=[ (900, 1000, "text_text_text value"), ], ), ], ), # text_checklist nested classification lb_types.TemporalClassificationQuestion( name="text_checklist", value=[ lb_types.TemporalClassificationAnswer( name="text_checklist_option_1", frames=[(1500, 2000)], classifications=[ lb_types.TemporalClassificationText( name="text_checklist_text", value=[ (1600, 2000, "text / checklist / option 1 / text classification value"), ], ), ], ), lb_types.TemporalClassificationAnswer( name="text_checklist_option_2", frames=[(1800, 2200)], ), ], ), # text_radio nested classification lb_types.TemporalClassificationQuestion( name="text_radio", value=[ lb_types.TemporalClassificationAnswer( name="text_radio_option_1", frames=[(2700, 3200)], classifications=[ lb_types.TemporalClassificationQuestion( name="text_radio_checklist", value=[ lb_types.TemporalClassificationAnswer( name="text_radio_checklist_option_1", frames=[(2800, 3100)], ), lb_types.TemporalClassificationAnswer( name="text_radio_checklist_option_3", frames=[(2900, 3200)], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="text_radio_option_2", frames=[(3201, 3500)], ), ], ), ], ) # RADIO_CLASS: Radio > Text/Radio/Checklist branches radio_class_annotation = lb_types.TemporalClassificationQuestion( name="radio_class", value=[ lb_types.TemporalClassificationAnswer( name="radio_class_option_1", frames=[ (200, 700), (1401, 2500), ], classifications=[ lb_types.TemporalClassificationText( name="radio_text", value=[ (300, 700, "radio_text value"), (1401, 2500, "radio_text value"), ], classifications=[ lb_types.TemporalClassificationQuestion( name="radio_text_checklist", value=[ lb_types.TemporalClassificationAnswer( name="option_1_radio_text_checklist", frames=[(1401, 2099)], ), lb_types.TemporalClassificationAnswer( name="option_2_radio_text_checklist", frames=[(1600, 2299)], ), lb_types.TemporalClassificationAnswer( name="option_3_radio_text_checklist", frames=[(1900, 2500)], ), ], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="radio_class_option_2", frames=[(701, 1400)], classifications=[ lb_types.TemporalClassificationQuestion( name="radio_radio", value=[ lb_types.TemporalClassificationAnswer( name="radio_radio_option_1", frames=[(701, 1199)], classifications=[ lb_types.TemporalClassificationText( name="radio_radio_text", value=[ (900, 1199, "radio_radio_text value"), ], ), ], ), lb_types.TemporalClassificationAnswer( name="radio_radio_option_2", frames=[(1200, 1400)], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="radio_class_option_3", frames=[(2600, 3500)], classifications=[ lb_types.TemporalClassificationQuestion( name="radio_checklist", value=[ lb_types.TemporalClassificationAnswer( name="radio_checklist_option_1", frames=[(2600, 3300)], classifications=[ lb_types.TemporalClassificationQuestion( name="radio_checklist_radio", value=[ lb_types.TemporalClassificationAnswer( name="radio_checklist_radio_option_1", frames=[(2600, 3300)], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="radio_checklist_option_2", frames=[(3000, 3500)], ), ], ), ], ), ], ) # CHECKLIST_CLASS: Checklist > Text/Radio/Checklist branches checklist_class_annotation = lb_types.TemporalClassificationQuestion( name="checklist_class", value=[ lb_types.TemporalClassificationAnswer( name="checklist_class_option_1", frames=[(200, 1899)], classifications=[ lb_types.TemporalClassificationText( name="checklist_text", value=[ (200, 1899, "checklist_text value"), ], classifications=[ lb_types.TemporalClassificationText( name="checklist_text_text", value=[ (400, 1899, "checklist_text_text value"), ], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="checklist_class_option_2", frames=[(900, 2500)], classifications=[ lb_types.TemporalClassificationQuestion( name="checklist_radio", value=[ lb_types.TemporalClassificationAnswer( name="option_1_checklist_radio", frames=[(900, 1999)], classifications=[ lb_types.TemporalClassificationQuestion( name="checklist_radio_radio", value=[ lb_types.TemporalClassificationAnswer( name="option_1_checklist_radio_radio", frames=[(1100, 1999)], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="option_2_checklist_radio", frames=[(2000, 2500)], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="checklist_class_option_3", frames=[(2600, 3500)], classifications=[ lb_types.TemporalClassificationQuestion( name="checklist_checklist", value=[ lb_types.TemporalClassificationAnswer( name="option_1_checklist_checklist", frames=[(2600, 3300)], classifications=[ lb_types.TemporalClassificationQuestion( name="checklist_checklist_checklist", value=[ lb_types.TemporalClassificationAnswer( name="option_2_checklist_checklist_checklist", frames=[(2600, 2999)], ), lb_types.TemporalClassificationAnswer( name="option_3_checklist_checklist_checklist", frames=[(2800, 3300)], ), ], ), ], ), lb_types.TemporalClassificationAnswer( name="option_2_checklist_checklist", frames=[(3000, 3500)], ), ], ), ], ), ], ) # Create Label with all temporal annotations label = Label( data={"global_key": asset["global_key"]}, annotations=[ text_class_annotation, radio_class_annotation, checklist_class_annotation, ], ) ndjson_predictions = list(NDLabel.from_common([label])) # TemporalNDJSON objects are returned directly (not wrapped in NDLabel.annotations) # They already have the correct structure: name, answer, dataRow predictions_for_upload = [pred.model_dump() for pred in ndjson_predictions] if args.upload_raw or args.upload_computed: print("\n" + "="*80) print("STARTING MAL UPLOAD PROCESS") print("="*80) # API Configuration api_key_local = "" api_key_prod = "" rest_endpoint_local = "http://localhost:3000/api/v1" endpoint_local = "http://localhost:8080/graphql" # Initialize client based on environment if args.env == 'prod': print(f"🌍 Using PROD environment") client = lb.Client(api_key=api_key_prod) else: print(f"🌍 Using LOCAL environment") client = lb.Client( api_key=api_key_local, endpoint=endpoint_local, rest_endpoint=rest_endpoint_local ) # Generate unique global key for the asset global_key_upload = f"audio-token-demo-class-based-{str(uuid.uuid4())}" asset_upload = { "row_data": "https://storage.googleapis.com/lb-artifacts-testing-public/audio/gpt_how_can_you_help_me.wav", "global_key": global_key_upload, "media_type": "AUDIO", } print(f"\n📦 Creating dataset...") dataset = client.create_dataset( name=f"audio_token_demo_class_based_dataset_{str(uuid.uuid4())[:8]}", iam_integration=None ) print(f"✓ Dataset created: {dataset.uid}") print(f"\n📤 Uploading data row with global_key: {global_key_upload}") task = dataset.create_data_rows([asset_upload]) task.wait_till_done() print(f"✓ Data row uploaded") print(f"\n🏗️ Creating ontology...") ontology = client.create_ontology( f"Audio with permutations class-based {str(uuid.uuid4())[:8]}", ontology_builder.asdict(), media_type=lb.MediaType.Audio, ) print(f"✓ Ontology created: {ontology.uid}") print(f"\n📋 Creating project...") project = client.create_project( name=f"Audio with permutations class-based {str(uuid.uuid4())[:8]}", media_type=lb.MediaType.Audio ) print(f"✓ Project created: {project.uid}") print(f"\n🔗 Connecting ontology to project...") project.connect_ontology(ontology) print(f"✓ Ontology connected") print(f"\n📦 Creating batch...") batch = project.create_batch( f"audio-token-batch-class-based-{str(uuid.uuid4())[:8]}", global_keys=[global_key_upload], priority=5, ) print(f"✓ Batch created: {batch.uid}") # Determine which predictions to upload print(f"\n📤 Preparing COMPUTED predictions from class-based API...") upload_predictions = predictions_for_upload # Update global_key in predictions to match uploaded asset for pred in upload_predictions: if 'dataRow' in pred and 'globalKey' in pred['dataRow']: pred['dataRow']['globalKey'] = global_key_upload upload_type = "COMPUTED" print(f"📊 Uploading {len(upload_predictions)} {upload_type} predictions...") try: upload_job = lb.MALPredictionImport.create_from_objects( client=client, project_id=project.uid, name=f"audio_token_mal_{upload_type.lower()}_{str(uuid.uuid4())[:8]}", predictions=upload_predictions, ) print("⏳ Waiting for MAL upload to complete...") upload_job.wait_till_done() if upload_job.errors: print(f"❌ Errors: {upload_job.errors}") else: print(f"✓ MAL upload completed successfully!") print(f"ℹ️ No errors reported") except Exception as e: print(f"❌ MAL upload failed: {e}") import traceback traceback.print_exc() print(f"\n🌐 View Project in Browser:") print(f" - Project ID: {project.uid}") print(f" - URL: http://localhost:3000/projects/{project.uid}/overview") print("="*80)