task.issues.add_file_issue("<text-about-issue>", ["<issue-tag-1>", "<issue-tag-2"])
task.issues.add_frame_issue(<frame-number>, "<text-about-issue>", ["<issue-tag-1>", "<issue-tag-2"])
task.issues.add_coordinate_issue(<frame-number>, <coordinate-x>, <coordinate-y>, "<text-about-issue>", ["<issue-tag-1>", "<issue-tag-2"])
Modality | File | Frame | Pinned/Coordinate |
---|---|---|---|
Images | ✅ | ❌ | ✅ |
Videos | ✅ | ✅ | ✅ |
Audio files | ✅ | ✅ | ❌ |
Text files | ✅ | ❌ | ❌ |
HTML files | ✅ | ❌ | ❌ |
PDFs | ✅ | ✅ | ✅ |
DICOM | ✅ | ✅ | ✅ |
NifTi | ✅ | ✅ | ✅ |
# Import dependencies
from encord import EncordUserClient
from encord.workflow import AnnotationStage, ReviewStage
from encord.objects import ChecklistAttribute, Object, ObjectInstance, Option, RadioAttribute, TextAttribute
from encord.objects.coordinates import BoundingBoxCoordinates
# User input
SSH_PATH = "/Users/chris-encord/ssh-private-key.txt"
PROJECT_ID = "ef4c2685-512a-4af9-9ac1-443f766c6c80"
# Authentication
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
ssh_private_key_path=SSH_PATH,
# For US platform users use "https://api.us.encord.com"
domain="https://api.encord.com",
)
project = user_client.get_project(PROJECT_ID)
workflow = project.workflow
ANNOTATE_ISSUES = [
{
"data_title": "cherries-010.jpg",
"file_issue": {"text": "Annotate file issue 1", "tags": ["incorrect object", "label too large"]},
"coordinate_issue": {"frame": 0, "x": 0.5, "y": 0.5, "text": "Annotate coordinate issue 1", "tags": ["incorrect object"]},
},
{
"data_title": "cherries-vid-001.mp4",
"file_issue": {"text": "Annotate file issue 1", "tags": ["incorrect object", "label too large"]},
"frame_issue": {"frame": 0, "text": "Annotate frame issue 1", "tags": ["incorrect object"]},
"coordinate_issue": {"frame": 0, "x": 0.5, "y": 0.5, "text": "Annotate coordinate issue 1", "tags": ["incorrect object", "label too large"]},
},
]
REVIEW_ISSUES = [
{
"data_title": "cherries-sequence",
"file_issue": {"text": "Review file issue 1", "tags": ["incorrect object"]},
"frame_issue": {"frame": 3, "text": "Review frame issue 1", "tags": ["label too large"]},
"coordinate_issue": {"frame": 0, "x": 0.3, "y": 0.3, "text": "Review coordinate issue 1", "tags": ["incorrect object", "label too large"]},
},
{
"data_title": "cherries-ig",
"file_issue": {"text": "Review file issue 1", "tags": ["incorrect object", "label too large"]},
"frame_issue": {"frame": 2, "text": "Review frame issue 1", "tags": ["incorrect object", "label too large"]},
"coordinate_issue": {"frame": 0, "x": 0.3, "y": 0.3, "text": "Review coordinate issue 1", "tags": ["incorrect object", "label too large"]},
},
{
"data_title": "cherries-004.jpg",
"file_issue": {"text": "Review file issue 1", "tags": ["incorrect object", "label too large"]},
"coordinate_issue": {"frame": 0, "x": 0.3, "y": 0.3, "text": "Review coordinate issue 1", "tags": ["incorrect object", "label too large"]},
},
]
# Annotate stage
annotate_stage = workflow.get_stage(name='Annotate 1', type_=AnnotationStage)
for issue in ANNOTATE_ISSUES:
data_title = issue["data_title"]
task = next((t for t in annotate_stage.get_tasks() if t.data_title == data_title), None)
if not task:
print(f"[Annotate] Task not found for: {data_title}")
continue
print(f"Processing Annotate Task: {data_title}")
task.assign('chris-encord@acme.com')
task.issues.add_file_issue(issue["file_issue"]["text"], issue["file_issue"]["tags"])
if "frame_issue" in issue:
fi = issue["frame_issue"]
task.issues.add_frame_issue(fi["frame"], fi["text"], fi["tags"])
ci = issue["coordinate_issue"]
task.issues.add_coordinate_issue(ci["frame"], ci["x"], ci["y"], ci["text"], ci["tags"])
label_row = project.list_label_rows_v2(data_hashes=[task.data_hash])[0]
label_row.initialise_labels()
box_object = label_row.ontology_structure.get_child_by_title("BoundingBox", type_=Object)
box_instance = box_object.create_instance()
box_instance.set_for_frames(
coordinates=BoundingBoxCoordinates(height=0.1, width=0.1, top_left_x=0.5, top_left_y=0.5),
)
label_row.add_object_instance(box_instance)
label_row.save()
task.submit()
# Review stage
review_stage = workflow.get_stage(name='Review 1', type_=ReviewStage)
for issue in REVIEW_ISSUES:
data_title = issue["data_title"]
task = next((t for t in review_stage.get_tasks() if t.data_title == data_title), None)
if not task:
print(f"[Review] Task not found for: {data_title}")
continue
print(f"Processing Review Task: {data_title}")
task.issues.add_file_issue(issue["file_issue"]["text"], issue["file_issue"]["tags"])
if "frame_issue" in issue:
fi = issue["frame_issue"]
task.issues.add_frame_issue(fi["frame"], fi["text"], fi["tags"])
ci = issue["coordinate_issue"]
task.issues.add_coordinate_issue(ci["frame"], ci["x"], ci["y"], ci["text"], ci["tags"])
for label_review in task.get_label_reviews():
label_review.reject(comment='Rejected due to issue', issue_tags=ci["tags"]) # Or use any tag set you prefer
task.reject()
Was this page helpful?