Project

import labelbox as lb
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
)

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>"
  # Optional: If not included, defaults to "None", which moves data rows to the "Done" bucket
  task_queue_id="<task_queue_id>"
)

# 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()

# Stream the export using a callback function
def json_stream_handler(output: labelbox.BufferedJsonConverterOutput):
  print(output.json)

export_task.get_buffered_stream(stream_type=labelbox.StreamType.RESULT).start(stream_handler=json_stream_handler)

# Collect all exported data into a list
export_json = [data_row.json for data_row in export_task.get_buffered_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()

# 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()

  # Stream the export using a callback function
  def json_stream_handler(output: labelbox.BufferedJsonConverterOutput):
    print(output.json)

  export_task.get_buffered_stream(stream_type=labelbox.StreamType.RESULT).start(stream_handler=json_stream_handler)

  # Collect all exported data into a list
  export_json = [data_row.json for data_row in export_task.get_buffered_stream()]

  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))

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}")