Project

import labelbox as lb
from labelbox.schema.quality_mode import QualityMode
client = lb.Client(api_key="<YOUR_API_KEY>")

# Create a new project
project = client.create_project(
    name="<project_name>",
    description="<project_description>",    # optional
    media_type=lb.MediaType.Image           # specify the media type
)

from labelbox.schema.quality_mode import QualityMode

project = client.create_project(
  name="benchmark-project",
  description="description",
  media_type=lb.MediaType.Image,
  quality_mode=QualityMode.Benchmark
)

from labelbox.schema.quality_mode import QualityMode

# create a project that uses consensus quality control
project = client.create_project(
  name="consensus-project",
  description="description",
  media_type=lb.MediaType.Image,
  quality_mode=QualityMode.Consensus
)

project = client.get_project("<project_id>")

# alternatively, you can get a dataset by name 
project = client.get_projects(where=lb.Project.name == "<project_name>").get_one()

project.create_batch(
  name="<unique_batch_name>",
  global_keys=["key1", "key2", "key3"],
  priority=5,
)

# if the project uses consensus, you can optionally supply a dictionary with consensus settings
# if provided, the batch will use consensus with the specificed coverage and votes
project.create_batch(
  name="<unique_batch_name>",
  data_rows=["<data_row_id>", "<data_row_id>"],
  priority=1,
  consensus_settings={"number_of_labels": 3, "coverage_percentage": 0.1}
)

task = project.create_batches(
  name_prefix="demo-create-batches-",
  global_keys=global_keys,
  priority=5
)

print("Errors: ", task.errors())
print("Result: ", task.result())

dataset = client.get_dataset("<dataset_id>")

task = project.create_batches_from_dataset(
    name_prefix="demo-dataset-",
    dataset_id=dataset.uid,
    priority=5
)

print("Errors: ", task.errors())
print("Result: ", task.result())

# get the batches (objects of the Batch class)
batches = project.batches()

# inspect one batch
next(batches)

# inspect all batches
for batch in batches:
  print(batch)
    
# for ease of use, you can convert the paginated collection to a list
list(batches)

project_id = "<project_id>"
batch_id = "<batch_id>"

# returns a Batch object
batch = client.get_batch(project_id, batch_id)

# the argument must be an object of the Ontology class
project.connect_ontology(ontology)

# get the members (objects of the ProjectMember class with relationships to a User and Role)  
members = project.members()

# inspect one member
member = next(members)
print(member.user(), member.role(), member.access_from)

# Display member info:
print(member.user().uid, member.user().email, member.role().name, access_from, sep="\t")

# inspect all members
for member in members:
  print(member.user(), member.role(), access_from)
  
# for ease of use, you can convert the paginated collection to a list
list(members)

# must be a PDF or HTML file
project.upsert_instructions("<local_file_path>")

# get the task queues (relationship to TaskQueue objects)
task_queues = project.task_queues()

# inspect all task queues
for task_queue in task_queues:
  print(task_queue)

project.move_data_rows_to_task_queue(
  data_row_ids=lb.GlobalKeys(["<global_key>", "<global_key>"]), # Use "lb.UniqueIds" for "<data_row_ids>"
  task_queue_id="<task_queue_id>" # Use "None" to move data rows to the "Done" bucket 
)

# Extract the global keys
export_params = {
    "data_row_details": True
}

export_task = project.export(params=export_params)

# Wait until the export task is complete
export_task.wait_till_done()

# Check for any errors in the export task
if export_task.has_errors():
  export_task.get_buffered_stream(
    stream_type=lb.StreamType.ERRORS
  ).start(stream_handler=lambda error: print(error))

export_json = export_task.get_stream()

global_keys = [item["data_row"]["global_key"] for item in export_json]

# Add LPOs
lpos = []
priority=1

# With global keys 
 for global_key in global_keys: 
   lpos.append((lb.GlobalKey(global_key), priority))
   priority+=1

# With data row ids
# data_row_ids = ["clw7jlmav35yn0768xrpawwrc", "clw7jlmav35yo0768a5amfztu"]
# for dr_id in data_row_ids: 
#   lpos.append((lb.UniqueId(dr_id), priority))
#   priority+=1

# With data row objects
# data_rows = [data_row_1, data_row_2]
# for data_row in data_rows: 
#  lpos.append((data_row, priority)) 
#  priority+=1

# Set data row priorities
project.set_labeling_parameter_overrides(lpos)

# Check results
project_lpos = list(project.labeling_parameter_overrides())
for lpo in project_lpos:
  print(lpo)

# Update LPOs

# With global keys
 global_keys = ["global_key1", "global_key2"]
 project.update_data_row_labeling_priority(data_rows=lb.GlobalKeys(global_keys), priority=1)

# With data row ids
# data_row_ids = ["clw7jlmav35yn0768xrpawwrc", "clw7jlmav35yo0768a5amfztu"]
# project.update_data_row_labeling_priority(data_rows=lb.UniqueIds(data_row_ids), priority=1)

# With data row objects
# data_rows = [data_row_1, data_row_2]
# project.update_data_row_labeling_priority(data_rows=data_rows, priority=1)


# Check results
project_lpos = list(project.labeling_parameter_overrides())

for lpo in project_lpos:
  print(lpo)

tags = project.update_project_resource_tags(["<project_tag_id>", "<project_tag_id>"])

tags = project.get_resource_tags()

# Retrieve the import jobs of a project
import_requests = project.bulk_import_requests()

# Retrieve a particular import request
job_id = "<job_id>"

import_request = [import_request for import_request 
 in project.bulk_import_requests()
 if import_request.uid == job_id][0]
 
# Delete an import and all associated pre annotations
# WARNING: this can't be undone
# import_request.delete()

# Example without details
overview = project.get_overview()

# Selection of some attributes
print(f"""
      To label:\t{overview.to_label / overview.total_data_rows:.2%}
      Labeled:\t{overview.labeled / overview.total_data_rows:.2%}
     """)
# To label:	18.37%
# Labeled:	81.63%


# Example with details
detailed_overview = project.get_overview(details=True)

# Task queues in review
print(f"""
      Number of data rows {detailed_overview.total_data_rows},
      In review:,
      \tQueues {detailed_overview.in_review["data"]},
      \tNumber of data rows  {detailed_overview.in_review["total"]},
      In rework:
      \tQueues {detailed_overview.in_rework["data"]},
      \tNumber of data rows  {detailed_overview.in_rework["total"]}
      """,
      sep="\n")
# Number of data rows 23220,
# In review:,
# Queues [{'Initial review task': 7}],
# Number of data rows  7,
# In rework:
#  Queues [{'Rework (all rejected)': 1830}],
#  Number of data rows  1830

# The return type of this method is an `ExportTask`, which is a wrapper of a`Task`
# Most of `Task` features are also present in `ExportTask`.
export_params= {
  "attachments": True,
  "metadata_fields": True,
  "data_row_details": True,
  "project_details": True,
  "label_details": True,
  "performance_details": True,
  "interpolated_frames": True
}

# Note: Filters follow AND logic, so typically using one filter is sufficient.
filters= {
  "last_activity_at": ["2000-01-01 00:00:00", "2050-01-01 00:00:00"],
  "label_created_at": ["2000-01-01 00:00:00", "2050-01-01 00:00:00"],
  "workflow_status": "InReview",
  "batch_ids": ["batch_id_1", "batch_id_2"],
  "data_row_ids": ["data_row_id_1", "data_row_id_2"],
  "global_keys": ["global_key_1", "global_key_2"]
}

export_task = project.export(params=export_params, filters=filters)
export_task.wait_till_done()

# Return a JSON output string from the export task results/errors one by one:
def json_stream_handler(output: lb.BufferedJsonConverterOutput):
  print(output.json)

if export_task.has_errors():
  export_task.get_buffered_stream(
    stream_type=lb.StreamType.ERRORS
  ).start(stream_handler=lambda error: print(error))

if export_task.has_result():
  export_json = export_task.get_buffered_stream(
    stream_type=lb.StreamType.RESULT
  ).start(stream_handler=json_stream_handler)

print("file size: ", export_task.get_total_file_size(stream_type=lb.StreamType.RESULT))
print("line count: ", export_task.get_total_lines(stream_type=lb.StreamType.RESULT))

# Set the export params to include/exclude certain fields. Make sure each of these fields are correctly grabbed 
export_params= {
  "attachments": True,
  "metadata_fields": True,
  "data_row_details": True,
  "project_details": True,
  "label_details": True,
  "performance_details": True
}

# You can set the range for last_activity_at and label_created_at. You can also set a list of data 
# row ids to export. 
# For context, last_activity_at captures the creation and modification of labels, metadata, status, comments and reviews.

# Note: Combinations of filters apply AND logic.
filters= {
  "last_activity_at": ["2000-01-01 00:00:00", "2050-01-01 00:00:00"],
  "label_created_at": ["2000-01-01 00:00:00", "2050-01-01 00:00:00"],
  "workflow_status": "InReview",
  "batch_ids": ["batch_id_1", "batch_id_2"],
  "data_row_ids": ["data_row_id_1", "data_row_id_2"],
  "global_keys": ["global_key_1", "global_key_2"]
}

export_task = project.export_v2(params=export_params, filters=filters)
export_task.wait_till_done()

if export_task.errors:
  print(export_task.errors)

export_json = export_task.result
print("results: ", export_json)

import requests

url = project.export_issues()
issues = requests.get(url).json()

# optionally, you can export only the open or resolved issues
open_issues_url = project.export_issues(status="Open")
resolved_issues_url = project.export_issues(status="Resolved")

cloned_project = project.clone()

project.update(name="<new_project_name>")

project.delete()

# name (str)
project.name

# description (str)
project.description

# updated at (datetime)
project.updated_at

# created at (datetime)
project.created_at

# last activity time (datetime)
project.last_activity_time

# number of required labels per consensus data row (int) 
project.auto_audit_number_of_labels

# default percentage of consensus data rows per batch (float)
project.auto_audit_percentage

# created by (relationship to User object)
user = project.created_by()

# organization (relationship to Organization object)
organization = project.organization()

# get the ontology connected to the project (relationship to Ontology object)
ontology = project.ontology()

# get the benchmarks (relationship to Benchmark objects)
benchmarks = project.benchmarks()

# inspect one benchmark
next(benchmarks)

# inspect all benchmarks
for benchmark in benchmarks:
  print(benchmark)
  
# for ease of use, you can convert the paginated collection to a list
list(benchmarks)

# get the webhooks connected to the project (relationship to Webhook objects)
webhooks = project.webhooks()

# inspect one webhook
next(webhooks)

# inspect all webhooks
for webhook in webhooks:
  print(webhook)
  
# for ease of use, you can convert the paginated collection to a list
list(webhooks)

# gets the LPOs created in the project (relationship to LabelingParameterOverride objects)
lpos = project.labeling_parameter_overrides()

# inspect one LPO
next(lpos)

# inspect all LPOs
for lpo in lpos:
  print(lpo)

# for ease of use, you can convert the paginated collection to a list
list(lpos)

# Get the data row id 
for lpo in lpos:
  print(lpo)
  print("Data row:", lpo.data_row().uid)

# Return the number of 
project.get_label_count()

from labelbox.schema.conflict_resolution_strategy import ConflictResolutionStrategy

send_to_annotate_params = {
    "source_project_id": project.uid,
    "annotations_ontology_mapping": annotation_ontology_mapping, # to be defined
    "exclude_data_rows_in_project": False,
    "override_existing_annotations_rule": ConflictResolutionStrategy.OverrideWithPredictions,
    "batch_priority": 5,
}

# Get task id to workflow you want to send data rows. If sent to initial labeling queue, labels will be pre-labels. 
queue_id = [queue.uid for queue in destination_project.task_queues() if queue.queue_type == "MANUAL_REVIEW_QUEUE" ][0]

task = client.send_to_annotate_from_catalog(
    destination_project_id=destination_project.uid,
    task_queue_id=queue_id, # ID of workflow task, set ID to None if you want to send data rows with labels to the Done queue.
    batch_name="Prediction Import Demo Batch",
    data_rows=lb.GlobalKeys(
        global_keys # Provide a list of global keys from source project
    ),
    params=send_to_annotate_params
    )

task.wait_till_done()

print(f"Errors: {task.errors}")