Skip to content

Rapidata flow item

RapidataFlowItem #

RapidataFlowItem(
    id: str,
    flow_id: str,
    openapi_service: OpenAPIService,
    flow_type: FlowType = "ranking",
)
Source code in src/rapidata/rapidata_client/flow/rapidata_flow_item.py 
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    id: str,
    flow_id: str,
    openapi_service: OpenAPIService,
    flow_type: FlowType = "ranking",
):
    self.id = id
    self.flow_id = flow_id
    self._openapi_service = openapi_service
    self._flow_type: FlowType = flow_type
    self._response_count: float | int | None = None

get_response_count #

get_response_count() -> float | int

Get the total number of responses for this flow item.

For ranking flow items the count is derived from the win/loss matrix by summing all entries, which triggers :meth:get_win_loss_matrix. For classify flow items it is read from the results via :meth:get_results. Either way this waits for the flow item to finish.

Returns:

Type Description
float | int

float | int: The total number of responses collected.
Source code in src/rapidata/rapidata_client/flow/rapidata_flow_item.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def get_response_count(self) -> float | int:
    """Get the total number of responses for this flow item.

    For ranking flow items the count is derived from the win/loss matrix by
    summing all entries, which triggers :meth:`get_win_loss_matrix`. For
    classify flow items it is read from the results via :meth:`get_results`.
    Either way this waits for the flow item to finish.

    Returns:
        float | int: The total number of responses collected.
    """
    with tracer.start_as_current_span("RapidataFlowItem.get_response_count"):
        if self._response_count is None:
            if self._flow_type == "ranking":
                self.get_win_loss_matrix()
            else:
                self.get_results()
        assert self._response_count is not None
        return self._response_count

get_status #

get_status() -> FlowItemState

Get the current state of this flow item.

Returns:

Name Type Description
FlowItemState FlowItemState

The current state (Pending, Running, Completed, Failed, Stopping, Stopped, or Incomplete).
Source code in src/rapidata/rapidata_client/flow/rapidata_flow_item.py 
57
58
59
60
61
62
63
64
65
66
def get_status(self) -> FlowItemState:
    """Get the current state of this flow item.

    Returns:
        FlowItemState: The current state (Pending, Running, Completed, Failed, Stopping, Stopped, or Incomplete).
    """
    with tracer.start_as_current_span("RapidataFlowItem.get_status"):
        logger.debug("Getting status for flow item '%s'", self.id)
        details = self._get_details()
        return details.state

get_results #

get_results() -> FlowItemResult | ClassifyFlowItemResult

Get the results of this flow item from the API.

Waits until the flow item has reached a terminal state.

Returns:

Name Type Description
FlowItemResult FlowItemResult | ClassifyFlowItemResult

For ranking flow items, a mapping of asset identifier to elo score and the total number of votes.
ClassifyFlowItemResult FlowItemResult | ClassifyFlowItemResult

For classify flow items, the majority category, response distribution and response count per asset identifier, and the total number of responses.
Source code in src/rapidata/rapidata_client/flow/rapidata_flow_item.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get_results(self) -> FlowItemResult | ClassifyFlowItemResult:
    """Get the results of this flow item from the API.

    Waits until the flow item has reached a terminal state.

    Returns:
        FlowItemResult: For ranking flow items, a mapping of asset identifier to elo score
            and the total number of votes.
        ClassifyFlowItemResult: For classify flow items, the majority category, response
            distribution and response count per asset identifier, and the total number of responses.
    """
    with tracer.start_as_current_span("RapidataFlowItem.get_results"):
        from rapidata.api_client.models.flow_item_state import FlowItemState

        logger.debug("Getting results for flow item '%s'", self.id)
        self._wait_for_state(
            target_states=[
                FlowItemState.COMPLETED,
                FlowItemState.FAILED,
                FlowItemState.STOPPED,
                FlowItemState.INCOMPLETE,
            ],
            check_interval=1,
            status_message="Flow item '%s' is in state %s, waiting for completion...",
        )

        if self._flow_type == "ranking":
            return self._get_ranking_results()
        return self._get_classify_results()

get_win_loss_matrix #

get_win_loss_matrix() -> DataFrame

Get the win/loss matrix of this ranking flow item from the API.

The win/loss matrix shows pairwise comparison counts where data[i][j] is the number of times row i was preferred over column j.

Returns:

Type Description
DataFrame

pd.DataFrame: A DataFrame where rows and columns are asset identifiers, and values are win/loss counts.

Raises:

Type Description
ValueError

If this is a classify flow item.
Source code in src/rapidata/rapidata_client/flow/rapidata_flow_item.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def get_win_loss_matrix(self) -> pd.DataFrame:
    """Get the win/loss matrix of this ranking flow item from the API.

    The win/loss matrix shows pairwise comparison counts where ``data[i][j]`` is
    the number of times row ``i`` was preferred over column ``j``.

    Returns:
        pd.DataFrame: A DataFrame where rows and columns are asset identifiers,
            and values are win/loss counts.

    Raises:
        ValueError: If this is a classify flow item.
    """
    with tracer.start_as_current_span("RapidataFlowItem.get_win_loss_matrix"):
        import pandas as pd
        from rapidata.api_client.models.flow_item_state import FlowItemState

        if self._flow_type != "ranking":
            raise ValueError(
                "The win/loss matrix is only available for ranking flow items."
            )

        logger.debug("Getting win/loss matrix for flow item '%s'", self.id)
        self._wait_for_state(
            target_states=[
                FlowItemState.COMPLETED,
                FlowItemState.FAILED,
                FlowItemState.STOPPED,
                FlowItemState.INCOMPLETE,
            ],
            check_interval=1,
            status_message="Flow item '%s' is in state %s, waiting for completion...",
        )

        result = self._openapi_service.flow.ranking_flow_item_api.flow_ranking_item_flow_item_id_vote_matrix_get(
            flow_item_id=self.id,
        )
        self._response_count = sum(sum(row) for row in result.data)

        return pd.DataFrame(
            data=result.data,
            index=pd.Index(result.index),
            columns=pd.Index(result.columns),
        )