Tabular Data Projects

from typing import Annotated
from pathlib import Path
import os

from encord_agents.tasks import Runner
from encord.objects.ontology_labels_impl import LabelRowV2
from encord.project import Project
from encord_agents.tasks.dependencies import dep_asset
from encord_agents.core.dependencies import Depends
from encord.objects.common import Shape

from tabular_utils import parse_csv_and_add_objects

# --- Configuration ---
ENCORD_SSH_KEY = "/Users/chris-encord/ssh-private-key.txt" # Replace with the file path to your SSH private key
PROJECT_HASH = "00000000-0000-0000-0000-000000000000" # Replace with unique Project ID of the tabular data Project
AGENT_STAGE = "Pre-label"
AGENT_PATHWAY = "Labelled"

# Inject into environment so Encord Agents can pick it up
os.environ["ENCORD_SSH_KEY_FILE"] = ENCORD_SSH_KEY

runner = Runner(project_hash=PROJECT_HASH)

@runner.stage(stage=AGENT_STAGE)
def agent_logic(
    lr: LabelRowV2, project: Project, asset: Annotated[Path, Depends(dep_asset)]
):
    ontology = project.ontology_structure
    text_object = ontology.objects[0]
    if text_object is None:
        raise Exception("No objects found")
    elif text_object.shape is not Shape.TEXT:
        raise Exception("Text object required")

    parse_csv_and_add_objects(text_object, lr, asset)

    return AGENT_PATHWAY

if __name__ == "__main__":
    runner.run()

import pandas as pd
from pathlib import Path

from encord.exceptions import OntologyError
from encord.objects.attributes import Attribute
from encord.objects.coordinates import TextCoordinates
from encord.objects.frames import Range
from encord.objects.ontology_labels_impl import LabelRowV2
from encord.objects.ontology_object import Object

def parse_csv_and_add_objects(
    text_object: Object, label_row: LabelRowV2, asset_link: Path
):
    csv_df = pd.read_csv(asset_link)
    filtered_columns = csv_df.columns

    read = None
    with asset_link as f:
        read = f.read_bytes()

    for index, row in csv_df.iloc[0:].iterrows():
        byte_range = get_byte_range_for_row(read, index + 1)
        if byte_range is None:
            raise Exception("Row does not exist")
        (start, end) = byte_range

        row_dict = row[filtered_columns].to_dict()

        add_text_object(text_object, label_row, Range(start=start, end=end), row_dict)

    label_row.save()


def get_byte_range_for_row(csv_content: bytes, row: int) -> tuple[int, int] | None:
    # Find all newline positions
    newline_positions = []
    current_pos = 0

    while True:
        newline_pos = csv_content.find(b"\n", current_pos)
        if newline_pos == -1:
            break
        newline_positions.append(newline_pos)
        current_pos = newline_pos + 1

    # Check if requested row exists
    if row > len(newline_positions):
        return None

    # Calculate start and end positions
    if row == 0:
        start_byte = 0
    else:
        start_byte = newline_positions[row - 1] + 1

    if row < len(newline_positions):
        end_byte = newline_positions[row]
    else:
        # Last row might not have a trailing newline
        end_byte = len(csv_content)

    return (start_byte, end_byte)


def add_text_object(
    text_object: Object,
    lr: LabelRowV2,
    byte_range: Range,
    csv_column_to_value: dict[str, str],
):
    new_instance = text_object.create_instance()
    new_instance.set_for_frames(coordinates=TextCoordinates(range=[byte_range]))
    for key, value in csv_column_to_value.items():
        try:
            attribute = text_object.get_child_by_title(title=key, type_=Attribute)
            if attribute.required:
                # If required attribute then this is the one they are writing so ignore
                continue
            new_instance.set_answer(attribute=attribute, answer=str(value))
        except OntologyError:
            # Ignore columns that don't have attributes
            pass
    lr.add_object_instance(new_instance)