Rapids manager

RapidsManager

RapidsManager(openapi_service: OpenAPIService)

Can be used to build different types of rapids. That can then be added to Validation sets

Source code in src/rapidata/rapidata_client/validation/rapids/rapids_manager.py

def __init__(self, openapi_service: OpenAPIService):
    self._openapi_service = openapi_service

classification_rapid

classification_rapid(
    instruction: str,
    answer_options: list[str],
    datapoint: str,
    truths: list[str],
    data_type: Literal["media", "text"] = "media",
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid

Build a classification rapid

Parameters:

Name	Type	Description	Default
instruction	str	The instruction/question to be shown to the labeler.	required
answer_options	list[str]	The options that the labeler can choose from to answer the question.	required
datapoint	str	The datapoint that the labeler will be labeling.	required
truths	list[str]	The correct answers to the question.	required
data_type	str	The type of the datapoint. Defaults to "media" (any form of image, video or audio).	'media'
context	str	The context is text that will be shown in addition to the instruction. Defaults to None.	None
media_context	list[str]	A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.	None
explanation	str	The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.	None

Source code in src/rapidata/rapidata_client/validation/rapids/rapids_manager.py

def classification_rapid(
    self,
    instruction: str,
    answer_options: list[str],
    datapoint: str,
    truths: list[str],
    data_type: Literal["media", "text"] = "media",
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid:
    """Build a classification rapid

    Args:
        instruction (str): The instruction/question to be shown to the labeler.
        answer_options (list[str]): The options that the labeler can choose from to answer the question.
        datapoint (str): The datapoint that the labeler will be labeling.
        truths (list[str]): The correct answers to the question.
        data_type (str, optional): The type of the datapoint. Defaults to "media" (any form of image, video or audio).
        context (str, optional): The context is text that will be shown in addition to the instruction. Defaults to None.
        media_context (list[str], optional): A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.
        explanation (str, optional): The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.
    """
    from rapidata.api_client.models.i_rapid_payload import IRapidPayload
    from rapidata.api_client.models.i_rapid_payload_classify_payload import (
        IRapidPayloadClassifyPayload,
    )
    from rapidata.api_client.models.classify_category import (
        ClassifyCategory,
    )
    from rapidata.api_client.models.i_validation_truth_model import (
        IValidationTruthModel,
    )
    from rapidata.api_client.models.i_validation_truth_model_attach_category_truth_model import (
        IValidationTruthModelAttachCategoryTruthModel,
    )
    from rapidata.rapidata_client.validation.rapids.rapids import Rapid

    if not isinstance(truths, list):
        raise ValueError("Truths must be a list of strings")

    if not all(truth in answer_options for truth in truths):
        raise ValueError("Truths must be part of the answer options")

    payload = IRapidPayload(
        actual_instance=IRapidPayloadClassifyPayload(
            _t="ClassifyPayload",
            categories=[
                ClassifyCategory(label=option, value=option)
                for option in answer_options
            ],
            title=instruction,
        )
    )
    model_truth = IValidationTruthModel(
        actual_instance=IValidationTruthModelAttachCategoryTruthModel(
            correctCategories=truths, _t="AttachCategoryTruth"
        )
    )

    return Rapid(
        asset=datapoint,
        data_type=data_type,
        context=context,
        media_context=media_context,
        explanation=explanation,
        payload=payload,
        truth=model_truth,
        random_correct_probability=len(truths) / len(answer_options),
    )

compare_rapid

compare_rapid(
    instruction: str,
    truth: str,
    datapoint: list[str],
    data_type: Literal["media", "text"] = "media",
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid

Build a compare rapid

Parameters:

Name	Type	Description	Default
instruction	str	The instruction that the labeler will be comparing the assets on.	required
truth	str	The correct answer to the comparison. (has to be one of the assets)	required
datapoint	list[str]	The two assets that the labeler will be comparing.	required
data_type	str	The type of the datapoint. Defaults to "media" (any form of image, video or audio).	'media'
context	str	The context is text that will be shown in addition to the instruction. Defaults to None.	None
media_context	list[str]	A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.	None
explanation	str	The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.	None

Source code in src/rapidata/rapidata_client/validation/rapids/rapids_manager.py

def compare_rapid(
    self,
    instruction: str,
    truth: str,
    datapoint: list[str],
    data_type: Literal["media", "text"] = "media",
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid:
    """Build a compare rapid

    Args:
        instruction (str): The instruction that the labeler will be comparing the assets on.
        truth (str): The correct answer to the comparison. (has to be one of the assets)
        datapoint (list[str]): The two assets that the labeler will be comparing.
        data_type (str, optional): The type of the datapoint. Defaults to "media" (any form of image, video or audio).
        context (str, optional): The context is text that will be shown in addition to the instruction. Defaults to None.
        media_context (list[str], optional): A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.
        explanation (str, optional): The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.
    """
    from rapidata.api_client.models.i_rapid_payload import IRapidPayload
    from rapidata.api_client.models.i_rapid_payload_compare_payload import (
        IRapidPayloadComparePayload,
    )
    from rapidata.api_client.models.i_validation_truth_model import (
        IValidationTruthModel,
    )
    from rapidata.api_client.models.i_validation_truth_model_compare_truth_model import (
        IValidationTruthModelCompareTruthModel,
    )
    from rapidata.rapidata_client.validation.rapids.rapids import Rapid

    if len(datapoint) != 2:
        raise ValueError("Compare rapid requires exactly two media paths")

    if truth not in datapoint:
        raise ValueError(
            f"Truth '{truth}' must be one of the datapoints: {datapoint}"
        )

    payload = IRapidPayload(
        actual_instance=IRapidPayloadComparePayload(
            _t="ComparePayload", criteria=instruction
        )
    )

    model_truth = IValidationTruthModel(
        actual_instance=IValidationTruthModelCompareTruthModel(
            _t="CompareTruth", winnerId=truth
        )
    )

    return Rapid(
        asset=datapoint,
        data_type=data_type,
        truth=model_truth,
        context=context,
        media_context=media_context,
        payload=payload,
        explanation=explanation,
        random_correct_probability=0.5,
    )

select_words_rapid

select_words_rapid(
    instruction: str,
    truths: list[int],
    datapoint: str,
    sentence: str,
    required_precision: float = 1,
    required_completeness: float = 1,
    explanation: str | None = None,
) -> Rapid

Build a select words rapid

Parameters:

Name	Type	Description	Default
instruction	str	The instruction for the labeler.	required
truths	list[int]	The indices of the words that are the correct answers.	required
datapoint	str	The asset that the labeler will be selecting words from.	required
sentence	str	The sentence that the labeler will be selecting words from. (split up by spaces)	required
required_precision	float	The required precision for the labeler to get the rapid correct (minimum ratio of the words selected that need to be correct). defaults to 1. (no wrong words can be selected)	1
required_completeness	float	The required completeness for the labeler to get the rapid correct (miminum ratio of total correct words selected). defaults to 1. (all correct words need to be selected)	1
explanation	str	The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.	None

Source code in src/rapidata/rapidata_client/validation/rapids/rapids_manager.py

def select_words_rapid(
    self,
    instruction: str,
    truths: list[int],
    datapoint: str,
    sentence: str,
    required_precision: float = 1,
    required_completeness: float = 1,
    explanation: str | None = None,
) -> Rapid:
    """Build a select words rapid

    Args:
        instruction (str): The instruction for the labeler.
        truths (list[int]): The indices of the words that are the correct answers.
        datapoint (str): The asset that the labeler will be selecting words from.
        sentence (str): The sentence that the labeler will be selecting words from. (split up by spaces)
        required_precision (float): The required precision for the labeler to get the rapid correct (minimum ratio of the words selected that need to be correct). defaults to 1. (no wrong words can be selected)
        required_completeness (float): The required completeness for the labeler to get the rapid correct (miminum ratio of total correct words selected). defaults to 1. (all correct words need to be selected)
        explanation (str, optional): The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.
    """
    from rapidata.api_client.models.i_rapid_payload import IRapidPayload
    from rapidata.api_client.models.i_rapid_payload_transcription_payload import (
        IRapidPayloadTranscriptionPayload,
    )
    from rapidata.api_client.models.transcription_word import TranscriptionWord
    from rapidata.api_client.models.transcription_truth_model_transcription_word import (
        TranscriptionTruthModelTranscriptionWord,
    )
    from rapidata.api_client.models.i_validation_truth_model import (
        IValidationTruthModel,
    )
    from rapidata.api_client.models.i_validation_truth_model_transcription_truth_model import (
        IValidationTruthModelTranscriptionTruthModel,
    )
    from rapidata.rapidata_client.validation.rapids.rapids import Rapid

    transcription_words = [
        TranscriptionWord(word=word, wordIndex=i)
        for i, word in enumerate(sentence.split(" "))
    ]

    correct_transcription_words: list[TranscriptionTruthModelTranscriptionWord] = []
    for index in truths:
        correct_transcription_words.append(
            TranscriptionTruthModelTranscriptionWord(
                word=transcription_words[index].word, wordIndex=index
            )
        )

    payload = IRapidPayload(
        actual_instance=IRapidPayloadTranscriptionPayload(
            _t="TranscriptionPayload",
            title=instruction,
            transcription=transcription_words,
        )
    )

    model_truth = IValidationTruthModel(
        actual_instance=IValidationTruthModelTranscriptionTruthModel(
            _t="TranscriptionTruth",
            correctWords=correct_transcription_words,
            requiredPrecision=required_precision,
            requiredCompleteness=required_completeness,
        )
    )

    return Rapid(
        payload=payload,
        truth=model_truth,
        asset=datapoint,
        sentence=sentence,
        explanation=explanation,
        random_correct_probability=len(correct_transcription_words)
        / len(transcription_words),
    )

locate_rapid

locate_rapid(
    instruction: str,
    truths: list[Box],
    datapoint: str,
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid

Build a locate rapid

Parameters:

Name	Type	Description	Default
instruction	str	The instruction on what the labeler should do.	required
truths	list[Box]	The bounding boxes of the object that the labeler ought to be locating.	required
datapoint	str	The asset that the labeler will be locating the object in.	required
context	str	The context is text that will be shown in addition to the instruction. Defaults to None.	None
media_context	list[str]	A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.	None
explanation	str	The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.	None

Source code in src/rapidata/rapidata_client/validation/rapids/rapids_manager.py

def locate_rapid(
    self,
    instruction: str,
    truths: list[Box],
    datapoint: str,
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid:
    """Build a locate rapid

    Args:
        instruction (str): The instruction on what the labeler should do.
        truths (list[Box]): The bounding boxes of the object that the labeler ought to be locating.
        datapoint (str): The asset that the labeler will be locating the object in.
        context (str, optional): The context is text that will be shown in addition to the instruction. Defaults to None.
        media_context (list[str], optional): A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.
        explanation (str, optional): The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.
    """
    from rapidata.api_client.models.i_rapid_payload import IRapidPayload
    from rapidata.api_client.models.i_rapid_payload_locate_payload import (
        IRapidPayloadLocatePayload,
    )
    from rapidata.api_client.models.i_validation_truth_model import (
        IValidationTruthModel,
    )
    from rapidata.api_client.models.i_validation_truth_model_locate_box_truth_model import (
        IValidationTruthModelLocateBoxTruthModel,
    )
    from rapidata.rapidata_client.validation.rapids.rapids import Rapid

    payload = IRapidPayload(
        actual_instance=IRapidPayloadLocatePayload(
            _t="LocatePayload", target=instruction
        )
    )

    model_truth = IValidationTruthModel(
        actual_instance=IValidationTruthModelLocateBoxTruthModel(
            _t="LocateBoxTruth",
            boundingBoxes=[truth.to_model() for truth in truths],
        )
    )

    coverage = self._calculate_boxes_coverage(
        truths,
    )

    return Rapid(
        payload=payload,
        truth=model_truth,
        asset=datapoint,
        context=context,
        media_context=media_context,
        explanation=explanation,
        random_correct_probability=coverage,
    )

draw_rapid

draw_rapid(
    instruction: str,
    truths: list[Box],
    datapoint: str,
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid

Build a draw rapid

Parameters:

Name	Type	Description	Default
instruction	str	The instructions on what the labeler	required
truths	list[Box]	The bounding boxes of the object that the labeler ought to be drawing. All boxes are attached as the ground truth — labelers are graded on whether their drawn lines fall within any of these boxes.	required
datapoint	str	The asset that the labeler will be drawing the object in.	required
context	str	The context is text that will be shown in addition to the instruction. Defaults to None.	None
media_context	list[str]	A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.	None
explanation	str	The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.	None

Source code in src/rapidata/rapidata_client/validation/rapids/rapids_manager.py

def draw_rapid(
    self,
    instruction: str,
    truths: list[Box],
    datapoint: str,
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid:
    """Build a draw rapid

    Args:
        instruction (str): The instructions on what the labeler
        truths (list[Box]): The bounding boxes of the object that the labeler ought to be drawing.
            All boxes are attached as the ground truth — labelers are graded on whether
            their drawn lines fall within any of these boxes.
        datapoint (str): The asset that the labeler will be drawing the object in.
        context (str, optional): The context is text that will be shown in addition to the instruction. Defaults to None.
        media_context (list[str], optional): A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.
        explanation (str, optional): The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.
    """
    from rapidata.api_client.models.i_rapid_payload import IRapidPayload
    from rapidata.api_client.models.i_rapid_payload_line_payload import (
        IRapidPayloadLinePayload,
    )
    from rapidata.api_client.models.i_validation_truth_model import (
        IValidationTruthModel,
    )
    from rapidata.api_client.models.i_validation_truth_model_locate_box_truth_model import (
        IValidationTruthModelLocateBoxTruthModel,
    )
    from rapidata.rapidata_client.validation.rapids.rapids import Rapid

    if not truths:
        raise ValueError("Draw rapid requires at least one truth bounding box")

    payload = IRapidPayload(
        actual_instance=IRapidPayloadLinePayload(
            _t="LinePayload", target=instruction
        )
    )

    model_truth = IValidationTruthModel(
        actual_instance=IValidationTruthModelLocateBoxTruthModel(
            _t="LocateBoxTruth",
            boundingBoxes=[truth.to_model() for truth in truths],
        )
    )

    coverage = self._calculate_boxes_coverage(
        truths,
    )

    return Rapid(
        payload=payload,
        truth=model_truth,
        asset=datapoint,
        context=context,
        media_context=media_context,
        explanation=explanation,
        random_correct_probability=coverage,
    )

timestamp_rapid

timestamp_rapid(
    instruction: str,
    truths: list[tuple[int, int]],
    datapoint: str,
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid

Build a timestamp rapid

Parameters:

Name	Type	Description	Default
instruction	str	The instruction for the labeler.	required
truths	list[tuple[int, int]]	The possible accepted timestamps intervals for the labeler (in miliseconds). The first element of the tuple is the start of the interval and the second element is the end of the interval.	required
datapoint	str	The asset that the labeler will be timestamping.	required
context	str	The context is text that will be shown in addition to the instruction. Defaults to None.	None
media_context	list[str]	A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.	None
explanation	str	The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.	None

Source code in src/rapidata/rapidata_client/validation/rapids/rapids_manager.py

def timestamp_rapid(
    self,
    instruction: str,
    truths: list[tuple[int, int]],
    datapoint: str,
    context: str | None = None,
    media_context: list[str] | None = None,
    explanation: str | None = None,
) -> Rapid:
    """Build a timestamp rapid

    Args:
        instruction (str): The instruction for the labeler.
        truths (list[tuple[int, int]]): The possible accepted timestamps intervals for the labeler (in miliseconds).
            The first element of the tuple is the start of the interval and the second element is the end of the interval.
        datapoint (str): The asset that the labeler will be timestamping.
        context (str, optional): The context is text that will be shown in addition to the instruction. Defaults to None.
        media_context (list[str], optional): A list of image URLs / paths that will be shown in addition to the instruction (can be combined with context). Pass a single-element list for one image, or multiple to display several images. Defaults to None.
        explanation (str, optional): The explanation that will be shown to the labeler if the answer is wrong. Defaults to None.
    """
    from rapidata.api_client.models.i_rapid_payload import IRapidPayload
    from rapidata.api_client.models.i_rapid_payload_scrub_payload import (
        IRapidPayloadScrubPayload,
    )
    from rapidata.api_client.models.scrub_truth_model_scrub_range import (
        ScrubTruthModelScrubRange,
    )
    from rapidata.api_client.models.i_validation_truth_model import (
        IValidationTruthModel,
    )
    from rapidata.api_client.models.i_validation_truth_model_scrub_truth_model import (
        IValidationTruthModelScrubTruthModel,
    )
    from rapidata.rapidata_client.validation.rapids.rapids import Rapid

    for truth in truths:
        if len(truth) != 2:
            raise ValueError(
                "The truths per datapoint must be a tuple of exactly two integers."
            )
        if truth[0] > truth[1]:
            raise ValueError(
                "The start of the interval must be smaller than the end of the interval."
            )

    payload = IRapidPayload(
        actual_instance=IRapidPayloadScrubPayload(
            _t="ScrubPayload", target=instruction
        )
    )

    model_truth = IValidationTruthModel(
        actual_instance=IValidationTruthModelScrubTruthModel(
            _t="ScrubTruth",
            validRanges=[
                ScrubTruthModelScrubRange(start=truth[0], end=truth[1])
                for truth in truths
            ],
        )
    )

    return Rapid(
        payload=payload,
        truth=model_truth,
        asset=datapoint,
        context=context,
        media_context=media_context,
        explanation=explanation,
        random_correct_probability=0.5,  # TODO: implement coverage ratio
    )