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