Getting Started with the Dataloop Python SDK (dtlpy) Tutorial

!pip install dtlpy --upgrade --quiet

import dtlpy as dl
import datetime

if dl.token_expired():
   dl.login() # Opens browser for login
   print(f"Logged in successfully to {dl.client_api.environment}")
else:
   print(f"Session active for {dl.client_api.info()['user_email']} in {dl.client_api.environment}")

# Define a unique name for your project
# Best practice: Use a combination of your username/initials and the purpose
user_email = dl.info()['user_email']
user_prefix = user_email.split('@')[0].replace('.', '').replace('-', '') # Simple prefix from email
project_name = f'{user_prefix}-sdk-getting-started'

# Check if the project exists, if not, create it
try:
    project = dl.projects.get(project_name=project_name)
    print(f"Successfully retrieved project: '{project.name}' (ID: {project.id})")
except dl.exceptions.NotFound:
    project = dl.projects.create(project_name=project_name)
    print(f"Successfully created project: '{project.name}' (ID: {project.id})")

# Print project details (optional)
# project.print()

# You can now access the project in the web interface
# project.open_in_web()

# You can list all projects you have access to
# print("\nListing projects accessible to you:")
# my_projects = dl.projects.list()
# my_projects.print()

# Define a unique name for your dataset within the project
dataset_name = 'my-sdk-dataset'

# Check if the dataset exists within the project
# if not, create it. Dataloop automatically creates a default Recipe and Ontology.
try:
    dataset = project.datasets.get(dataset_name=dataset_name)
    print(f"Successfully retrieved dataset: '{dataset.name}' (ID: {dataset.id})")
except dl.exceptions.NotFound:
    dataset = project.datasets.create(dataset_name=dataset_name)
    print(f"Successfully created dataset: '{dataset.name}' (ID: {dataset.id}) in project '{project.name}'.")

# Print dataset details (optional)
# dataset.print()

# List datasets within the project we retrieved/created
# print(f"\nListing datasets in project '{project.name}'...")
# datasets = project.datasets.list()
# datasets.print()

# Retrieve the dataset's default recipe
# Datasets usually have one recipe by default upon creation
try:
    recipe = dataset.recipes.list()[0] # Get the first recipe (usually the default)
    print(f"Retrieved recipe: '{recipe.title}' (ID: {recipe.id})")
except IndexError:
    # This should ideally not happen for SDK-created datasets, but as a fallback:
    print("No default recipe found. Creating one...")
    recipe = dataset.recipes.create()
    dataset.metadata['system']['recipes'] = [recipe.id] # Link recipe to dataset
    dataset.update(system_metadata=True)
    print(f"Created and linked recipe: '{recipe.title}' (ID: {recipe.id})")

# Retrieve the ontology linked to the recipe
# Recipes usually have one ontology by default
try:
    ontology = recipe.ontologies.list()[0] # Get the first ontology
    print(f"Retrieved ontology: '{ontology.title}' (ID: {ontology.id}) with {len(ontology.labels)} labels initially.")
except IndexError:
    # This should also ideally not happen.
    print("No default ontology found for the recipe. Creating one...")
    ontology = recipe.ontologies.create(labels=[]) # Create empty ontology
    recipe.ontology_ids = [ontology.id] # Link ontology to recipe
    recipe.update()
    print(f"Created and linked ontology: '{ontology.title}' (ID: {ontology.id})")

# Define the labels we want to use later for annotation
labels_to_add = {
    "Object1": '#FF0000',  # Red
    "CornerPoint": '#00FF00', # Lime Green
    "IndoorScene": '#0000FF', # Blue
    "MyTextLabel": '#FFFF00' # Yellow (For potential text annotation)
}

# Add labels to the ontology if they don't exist
labels_updated = False
existing_labels = [lbl.tag for lbl in ontology.labels]

for label_name, color_hex in labels_to_add.items():
    if label_name not in existing_labels:
        print(f"Adding label '{label_name}'...")
        ontology.add_label(label_name=label_name, color=color_hex)
        labels_updated = True
    else:
        print(f"Label '{label_name}' already exists.")

# Add a simple label first, then we'll add attributes to it separately
attributes_label_name = 'AttributedObject'
if attributes_label_name not in existing_labels:
    print(f"Adding label '{attributes_label_name}'...")
    ontology.add_label(label_name=attributes_label_name, color='#FFA500')  # Orange
    labels_updated = True

# Update the ontology on the platform *only* if changes were made
if labels_updated:
    ontology = ontology.update()
    print(f"Ontology updated successfully. It now has {len(ontology.labels)} labels.")
else:
    print("No new labels needed to be added.")

# Now add attributes using the correct method
print("Adding attributes to the ontology...")

# Add a boolean attribute (Yes/No)
ontology.update_attributes(
    key='occluded',
    title='Is Occluded?',
    attribute_type=dl.AttributesTypes.YES_NO,
    scope=[attributes_label_name]  # Apply only to the AttributedObject label
)

# Add a multi-choice attribute (Radio Button)
ontology.update_attributes(
    key='size',
    title='Object Size',
    attribute_type=dl.AttributesTypes.RADIO_BUTTON,
    values=['small', 'medium', 'large'],
    scope=[attributes_label_name]  # Apply only to the AttributedObject label
)

print("Attributes added successfully!")

# Optional: Print ontology labels
print("\nCurrent Ontology Labels:")
ontology.print()

# Example shortcut:
# dataset.add_label(label_name='ShortcutLabel', color='#FFC0CB') 

import os
from PIL import Image
import numpy as np

data_dir = 'data/getting_started_with_sdk'
image_filename = 'sample.jpg'
text_filename = 'sample.txt'
image_filepath = os.path.join(data_dir, image_filename)
text_filepath = os.path.join(data_dir, text_filename)

# Create directory if it doesn't exist
os.makedirs(data_dir, exist_ok=True)

# Create dummy image if it doesn't exist
if not os.path.exists(image_filepath):
    print(f"Creating dummy image at: {image_filepath}")
    dummy_image = Image.fromarray(np.random.randint(0, 256, (100, 150, 3), dtype=np.uint8))
    dummy_image.save(image_filepath)
else:
    print(f"Image already exists at: {image_filepath}")

# Create dummy text file if it doesn't exist
if not os.path.exists(text_filepath):
    print(f"Creating dummy text file at: {text_filepath}")
    with open(text_filepath, 'w') as f:
        f.write("This is a sample text file for the Dataloop SDK getting started guide.")
else:
    print(f"Text file already exists at: {text_filepath}")

import os

data_dir = 'data/getting_started_with_sdk'
image_filename = 'sample.jpg' # Ensure this file exists in the 'data' directory
text_filename = 'sample.txt' # Ensure this file exists in the 'data' directory
image_filepath = os.path.join(data_dir, image_filename)
text_filepath = os.path.join(data_dir, text_filename)

# Check if files exist before uploading
if not os.path.exists(image_filepath):
    raise FileNotFoundError(f"Image file not found: {image_filepath}. Please create it.")
if not os.path.exists(text_filepath):
    raise FileNotFoundError(f"Text file not found: {text_filepath}. Please create it.")

# Upload the local files to the dataset
# 'remote_path' specifies the directory structure within the Dataloop dataset
print("Uploading items...")

# Use overwrite=True to avoid errors if the item already exists
image_item: dl.Item = dataset.items.upload(
    local_path=image_filepath,
    remote_path='/images',
    overwrite=True
    )

print(f"Uploaded/Retrieved image item: {image_item.filename}, ID: {image_item.id}")
# image_item.print()

text_item: dl.Item = dataset.items.upload(
    local_path=text_filepath,
    remote_path='/docs',
    overwrite=True
    )

print(f"\nUploaded/Retrieved text item: {text_item.filename}, ID: {text_item.id}")
# text_item.print()

# List items in the dataset (can use filters)
print("\nListing all items in the dataset:")
all_items_page = dataset.items.list()
all_items_page.print()
print(f"Total items in dataset: {all_items_page.items_count}")

print("\nListing items in the '/images' directory:")
filters = dl.Filters(resource=dl.FiltersResource.ITEM)
filters.add(field='dir', values='/images') # Use add() method
image_items_page = dataset.items.list(filters=filters)
image_items_page.print()
print(f"Items in /images: {image_items_page.items_count}")

# Make sure we have the item IDs from the upload step
image_item_id = image_item.id
text_item_id = text_item.id
image_remote_filepath = image_item.filename # Get the actual remote path
text_remote_filepath = text_item.filename # Get the actual remote path

# Get an item by its path
try:
    retrieved_image_item = dataset.items.get(filepath=image_remote_filepath)
    print(f"\nRetrieved image item by path: {retrieved_image_item.filename}")
    # retrieved_image_item.print()
except dl.exceptions.NotFound:
    print(f"Error: Could not find item by path: {image_remote_filepath}")

# Get an item by its item id
try:
    retrieved_text_item = dataset.items.get(item_id=text_item_id)
    print(f"\nRetrieved text item by item id: {retrieved_text_item.id}")
    # retrieved_text_item.print()
except dl.exceptions.NotFound:
     print(f"Error: Could not find item by ID: {text_item_id}")

import json

# Make sure we successfully retrieved the image item
if 'retrieved_image_item' in locals():
    # Print existing metadata
    print("\nItem System Metadata:")
    print(json.dumps(retrieved_image_item.system, indent=2))
    print("\nItem User Metadata (before update):")
    print(json.dumps(retrieved_image_item.metadata.get('user', {}), indent=2))

    # Add/Update user metadata
    # The metadata is a dictionary. We usually add custom info under the 'user' key.

    # Ensure 'user' key exists and is a dictionary
    if 'user' not in retrieved_image_item.metadata or not isinstance(retrieved_image_item.metadata['user'], dict):
        retrieved_image_item.metadata['user'] = {}
    retrieved_image_item.metadata['user']['processed_by_script'] = True
    retrieved_image_item.metadata['user']['custom_info'] = 'This is from the SDK notebook'
    retrieved_image_item.metadata['user']['run_timestamp'] = datetime.datetime.now(datetime.timezone.utc).isoformat()

    # Push the update to the platform
    retrieved_image_item = retrieved_image_item.update()
    print("\nItem User Metadata (after update):")
    print(json.dumps(retrieved_image_item.metadata.get('user', {}), indent=2))
else:
    print("\nSkipping metadata update because image item retrieval failed.")

download_dir = 'downloaded_data'
os.makedirs(download_dir, exist_ok=True)

if 'retrieved_image_item' in locals():
    download_path = os.path.join(download_dir, retrieved_image_item.name)
    filepath = retrieved_image_item.download(local_path=download_path)
    print(f"Item '{retrieved_image_item.name}' downloaded successfully to: {filepath}")
else:
    print("Skipping item download because image item retrieval failed.")

# Ensure the image item was retrieved before creating the prompt item
if 'retrieved_image_item' in locals():
    # Create a PromptItem object (this doesn't upload yet)
    # The name will be the filename in the platform.
    prompt_item_name = "my-first-prompt.json"
    prompt_item_obj = dl.PromptItem(name=prompt_item_name)

    # Create a prompt (a single turn in a conversation)
    # Use a unique key for each prompt within the item
    user_prompt_key = "user_query_1"
    user_prompt = dl.Prompt(key=user_prompt_key, role="user")  # Default role is 'user'

    # Add elements to the prompt
    user_prompt.add_element(
        mimetype=dl.PromptType.TEXT,
        value="Describe this image:"
    )
    # Reference the image item we uploaded earlier by its data stream
    user_prompt.add_element(
        mimetype=dl.PromptType.IMAGE,
        value=retrieved_image_item.stream,
    )

    # Add the prompt to the PromptItem
    prompt_item_obj.prompts.append(user_prompt)

    # We can upload the prompt item directly
    uploaded_prompt_item: dl.Item = dataset.items.upload(
        local_path=prompt_item_obj,
        remote_path='/prompts',
        overwrite=True
        )
    print(f"Uploaded prompt item: {uploaded_prompt_item.filename} (ID: {uploaded_prompt_item.id})")

    # Open the item in the web interface
    # uploaded_prompt_item.open_in_web() # Uncomment to open
else:
    print("Skipping Prompt Item creation because image item retrieval failed.")

# Let's add annotations to the image item we uploaded
if 'retrieved_image_item' in locals():
    item_to_annotate = retrieved_image_item
    print(f"Preparing to annotate item: {item_to_annotate.name}")

    # Use the Annotation Builder for the specific item
    builder = item_to_annotate.annotations.builder()

    # Add a Bounding Box annotation
    # Coordinates are relative to the top-left corner (0,0)
    builder.add(
        annotation_definition=dl.Box(
            top=10,
            left=10,
            bottom=50, # Increased size slightly
            right=60,
            label="Object1" # Ensure this label exists in the ontology
        ),
        object_id=1 # Optional: for tracking instances
    )

    # Add a Point annotation
    builder.add(
        annotation_definition=dl.Point(
            x=75,
            y=75,
            label="CornerPoint" # Ensure this label exists in the ontology
        ),
         object_id=2 # Optional: for tracking instances
    )

    # Add a Classification annotation (no geometry)
    builder.add(
        annotation_definition=dl.Classification(label="IndoorScene") # Ensure this label exists in the ontology
    )

    # The builder now holds these annotations in memory.
    # Before uploading, you might want to clear existing annotations for this item
    # Be cautious with this in real projects!
    # item_to_annotate.annotations.delete(filters=None) # Deletes ALL annotations on the item

    # Upload them to the platform:
    print("Uploading annotations...")
    try:
        uploaded_annotations = item_to_annotate.annotations.upload(annotations=builder)
        print(f"Successfully uploaded {len(uploaded_annotations)} annotations.")

        # You can retrieve the annotations on the item again to verify
        annotations_on_item = item_to_annotate.annotations.list()
        print("\nAnnotations now on the item:")
        annotations_on_item.print() # Prints a summary table

        # Open the item in the web interface
        # item_to_annotate.open_in_web() # Uncomment to open
    except Exception as e:
        print(f"Error uploading annotations: {e}")
        print("Ensure the labels used exist in the dataset's ontology (Section 4).")
else:
    print("Skipping annotation creation because image item retrieval failed.")

# Make sure we have a valid item and user email
assignee_email = dl.info()['user_email'] # Assign task to yourself
if 'image_item' in locals():
    # Define due date (e.g., 1 week from now)
    due_date = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=7)

    # Create an annotation task for the image item
    task_name = f"SDK-GettingStarted-Task-{datetime.datetime.now().strftime('%Y%m%d-%H%M')}"
    print(f"Creating task '{task_name}'...")
    try:
        task: dl.Task = dataset.tasks.create(
            task_name=task_name,
            due_date=due_date.timestamp(),
            assignee_ids=[assignee_email],
            # You can assign specific items or use filters
            items=[image_item], # Assign the specific image item we uploaded
            # Example using filters to assign all items in '/images':
            # filters=dl.Filters(resource=dl.FiltersResource.ITEM, field='dir', values='/images')
        )
        print(f"Task created successfully: {task.name} (ID: {task.id})")

        # Creating a task automatically creates assignments for the assignees
        # You can list assignments for the task
        assignments = task.assignments.list()
        print("\nAssignments created for the task:")
        assignments.print()
    except Exception as e:
        print(f"Error creating task: {e}")
else:
    print("Skipping task creation because image item upload/retrieval failed.")