Skip to content

Editor Agents Reference

GCP

encord_agents.gcp.dependencies

Dependencies for injection in GCP Cloud run functions.

This module contains dependencies that you can inject within your cloud functions. Dependencies that depend on others don't need to be used together. They'll work just fine alone.

Note that you can also use the following typed parameters. If the type annotations are not present, the injection mechanism cannot resolve the them:

from encord.project import Project
from encord.objects.ontology_labels_impl import LabelRowV2
from encord_agents import FrameData
...
@app.post("/my-agent-route")
def my_agent(
    frame_data: FrameData,
    project: Project,
    label_row: LabelRowV2,
):
    ...
  • FrameData is automatically injected via the api request body.
  • Project is automatically loaded based on the frame data.
  • label_row_v2 is automatically loaded based on the frame data.

dep_client 

dep_client() -> EncordUserClient

Dependency to provide an authenticated user client.

Example:

from encord.user_client import EncordUserClient
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_client
...
@editor_agent()
def (
    client: Annotated[EncordUserClient, Depends(dep_client)]
):
    # Client will authenticated and ready to use.
    client.get_dataset("")
Source code in encord_agents/gcp/dependencies.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def dep_client() -> EncordUserClient:
    """
    Dependency to provide an authenticated user client.

    **Example:**

    ```python
    from encord.user_client import EncordUserClient
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_client
    ...
    @editor_agent()
    def (
        client: Annotated[EncordUserClient, Depends(dep_client)]
    ):
        # Client will authenticated and ready to use.
        client.get_dataset("")
    ```

    """
    return get_user_client()

dep_data_lookup 

dep_data_lookup(lookup: Annotated[DataLookup, Depends(DataLookup.sharable)]) -> DataLookup

Get a lookup to easily retrieve data rows and storage items associated with the given task.

Info

If you're just looking to get the associated storage item to a task, consider using dep_storage_item instead.

The lookup can, e.g., be useful for

  • Updating client metadata
  • Downloading data from signed urls
  • Matching data to other projects

Example:

from typing_extensions import Annotated
from encord.storage import StorageItem
from encord_agents import FrameData
from encord_agents.gcp import editor_agent, Depends
from encord_agents.gcp.dependencies import DataLookup, dep_data_lookup

@editor_agent()
def my_agent(
    frame_data: FrameData,
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
):
    print("data hash", lookup.get_data_row(frame_data.data_hash))
    print("storage item", lookup.get_storage_item(frame_data.data_hash))
    ...

Parameters:

  • lookup (Annotated[DataLookup, Depends(sharable)]) –

    The object that you can use to lookup data rows and storage items. Automatically injected.

Returns:

  • DataLookup –

    The (shared) lookup object.

Source code in encord_agents/gcp/dependencies.py 
146
147
148
149
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
def dep_data_lookup(lookup: Annotated[DataLookup, Depends(DataLookup.sharable)]) -> DataLookup:
    """
    Get a lookup to easily retrieve data rows and storage items associated with the given task.

    !!! info
        If you're just looking to get the associated storage item to a task, consider using `dep_storage_item` instead.


    The lookup can, e.g., be useful for

    * Updating client metadata
    * Downloading data from signed urls
    * Matching data to other projects

    **Example:**

    ```python
    from typing_extensions import Annotated
    from encord.storage import StorageItem
    from encord_agents import FrameData
    from encord_agents.gcp import editor_agent, Depends
    from encord_agents.gcp.dependencies import DataLookup, dep_data_lookup

    @editor_agent()
    def my_agent(
        frame_data: FrameData,
        lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
    ):
        print("data hash", lookup.get_data_row(frame_data.data_hash))
        print("storage item", lookup.get_storage_item(frame_data.data_hash))
        ...


    ```


    Args:
        lookup: The object that you can use to lookup data rows and storage items. Automatically injected.

    Returns:
        The (shared) lookup object.

    """
    return lookup

dep_object_crops 

dep_object_crops(filter_ontology_objects: list[Object | str] | None = None) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]

Get a list of object instances and frame crops associated with each object.

Useful, e.g., to be able to run each crop against a model.

Example:

@editor_agent
def my_agent(crops: Annotated[list[InstanceCrop], Depends[dep_object_crops(filter_ontology_objects=["eBw/75bg"])]]):
    for crop in crops:
        crop.content  # <- this is raw numpy rgb values
        crop.frame    # <- this is the frame number in video
        crop.instance # <- this is the obejct instance from the label row
    ...

Parameters:

  • filter_ontology_objects (list[Object | str] | None, default: None ) –

    Specify a list of ontology objects to include.

Returns: The dependency to be injected into the cloud function.

Source code in encord_agents/gcp/dependencies.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def dep_object_crops(
    filter_ontology_objects: list[Object | str] | None = None,
) -> Callable[[FrameData, LabelRowV2, NDArray[np.uint8]], list[InstanceCrop]]:
    """
    Get a list of object instances and frame crops associated with each object.

    Useful, e.g., to be able to run each crop against a model.

    **Example:**

    ```python
    @editor_agent
    def my_agent(crops: Annotated[list[InstanceCrop], Depends[dep_object_crops(filter_ontology_objects=["eBw/75bg"])]]):
        for crop in crops:
            crop.content  # <- this is raw numpy rgb values
            crop.frame    # <- this is the frame number in video
            crop.instance # <- this is the obejct instance from the label row
        ...
    ```

    Args:
        filter_ontology_objects: Specify a list of ontology objects to include.

    Returns: The dependency to be injected into the cloud function.

    """
    legal_feature_hashes = {
        o.feature_node_hash if isinstance(o, Object) else o for o in (filter_ontology_objects or [])
    }

    def _dep_object_crops(
        frame_data: FrameData, lr: LabelRowV2, frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ) -> list[InstanceCrop]:
        legal_shapes = {Shape.POLYGON, Shape.BOUNDING_BOX, Shape.ROTATABLE_BOUNDING_BOX, Shape.BITMASK}
        return [
            InstanceCrop(
                frame=frame_data.frame,
                content=crop_to_object(frame, o.get_annotation(frame=frame_data.frame).coordinates),  # type: ignore
                instance=o,
            )
            for o in lr.get_object_instances(filter_frames=frame_data.frame)
            if o.ontology_item.shape in legal_shapes
            and (not legal_feature_hashes or o.feature_hash in legal_feature_hashes)
        ]

    return _dep_object_crops

dep_single_frame 

dep_single_frame(lr: LabelRowV2) -> NDArray[np.uint8]

Dependency to inject the first frame of the underlying asset.

The downloaded asset will be named lr.data_hash.{suffix}. When the function has finished, the downloaded file will be removed from the file system.

Example:

from encord_agents import FrameData
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_single_frame
...

@editor_agent()
def my_agent(
    frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
):
    assert frame.ndim == 3, "Will work"

Parameters:

  • lr (LabelRowV2) –

    The label row. Automatically injected (see example above).

Returns:

  • NDArray[uint8] –

    Numpy array of shape [h, w, 3] RGB colors.

Source code in encord_agents/gcp/dependencies.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def dep_single_frame(lr: LabelRowV2) -> NDArray[np.uint8]:
    """
    Dependency to inject the first frame of the underlying asset.

    The downloaded asset will be named `lr.data_hash.{suffix}`.
    When the function has finished, the downloaded file will be removed from the file system.

    **Example:**

    ```python
    from encord_agents import FrameData
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_single_frame
    ...

    @editor_agent()
    def my_agent(
        frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ):
        assert frame.ndim == 3, "Will work"
    ```

    Args:
        lr: The label row. Automatically injected (see example above).

    Returns:
        Numpy array of shape [h, w, 3] RGB colors.

    """
    with download_asset(lr, frame=0) as asset:
        img = cv2.cvtColor(cv2.imread(asset.as_posix()), cv2.COLOR_BGR2RGB)

    return np.asarray(img, dtype=np.uint8)

dep_storage_item 

dep_storage_item(lookup: Annotated[DataLookup, Depends(dep_data_lookup)], frame_data: FrameData) -> StorageItem

Get the storage item associated with the underlying agent task.

The StorageItem is useful for multiple things like

  • Updating client metadata
  • Reading file properties like storage location, fps, duration, DICOM tags, etc.

Example

from typing_extensions import Annotated
from encord.storage import StorageItem
from encord_agents.gcp import editor_agent, Depends
from encord_agents.gcp.dependencies import dep_storage_item


@editor_agent()
def my_agent(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]):
    print("uuid", storage_item.uuid)
    print("client_metadata", storage_item.client_metadata)
    ...
Source code in encord_agents/gcp/dependencies.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def dep_storage_item(
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)],
    frame_data: FrameData,
) -> StorageItem:
    r"""
    Get the storage item associated with the underlying agent task.

    The [`StorageItem`](https://docs.encord.com/sdk-documentation/sdk-references/StorageItem){ target="\_blank", rel="noopener noreferrer" }
    is useful for multiple things like

    * Updating client metadata
    * Reading file properties like storage location, fps, duration, DICOM tags, etc.

    **Example**

    ```python
    from typing_extensions import Annotated
    from encord.storage import StorageItem
    from encord_agents.gcp import editor_agent, Depends
    from encord_agents.gcp.dependencies import dep_storage_item


    @editor_agent()
    def my_agent(storage_item: Annotated[StorageItem, Depends(dep_storage_item)]):
        print("uuid", storage_item.uuid)
        print("client_metadata", storage_item.client_metadata)
        ...
    ```

    """
    return lookup.get_storage_item(frame_data.data_hash)

dep_video_iterator 

dep_video_iterator(lr: LabelRowV2) -> Generator[Iterator[Frame], None, None]

Dependency to inject a video frame iterator for doing things over many frames.

Intended use

from encord_agents import FrameData
from encord_agents.gcp import editor_agent
from encord_agents.gcp.dependencies import dep_video_iterator
...

@editor_agent()
def my_agent(
    video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
):
    for frame in video_frames:
        print(frame.frame, frame.content.shape)

Parameters:

  • lr (LabelRowV2) –

    Automatically injected label row dependency.

Raises:

  • NotImplementedError –

    Will fail for other data types than video.

Yields:

  • Iterator[Frame] –

    An iterator.

Source code in encord_agents/gcp/dependencies.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def dep_video_iterator(lr: LabelRowV2) -> Generator[Iterator[Frame], None, None]:
    """
    Dependency to inject a video frame iterator for doing things over many frames.

    **Intended use**

    ```python
    from encord_agents import FrameData
    from encord_agents.gcp import editor_agent
    from encord_agents.gcp.dependencies import dep_video_iterator
    ...

    @editor_agent()
    def my_agent(
        video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
    ):
        for frame in video_frames:
            print(frame.frame, frame.content.shape)
    ```

    Args:
        lr: Automatically injected label row dependency.

    Raises:
        NotImplementedError: Will fail for other data types than video.

    Yields:
        An iterator.

    """
    if not lr.data_type == DataType.VIDEO:
        raise NotImplementedError("`dep_video_iterator` only supported for video label rows")

    with download_asset(lr, None) as asset:
        yield iter_video(asset)

encord_agents.gcp.wrappers

editor_agent 

editor_agent() -> Callable[[AgentFunction], Callable[[Request], Response]]

Wrapper to make resources available for gcp editor agents.

The editor agents are intended to be used via dependency injections. You can learn more via out documentation.

Source code in encord_agents/gcp/wrappers.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def editor_agent() -> Callable[[AgentFunction], Callable[[Request], Response]]:
    """
    Wrapper to make resources available for gcp editor agents.

    The editor agents are intended to be used via dependency injections.
    You can learn more via out [documentation](https://agents-docs.encord.com).
    """

    def context_wrapper_inner(func: AgentFunction) -> Callable:
        dependant = get_dependant(func=func)

        @wraps(func)
        def wrapper(request: Request) -> Response:
            frame_data = FrameData.model_validate_json(orjson.dumps(request.form.to_dict()))
            logging.info(f"Request: {frame_data}")

            client = get_user_client()
            project = client.get_project(str(frame_data.project_hash))

            label_row: LabelRowV2 | None = None
            if dependant.needs_label_row:
                label_row = project.list_label_rows_v2(data_hashes=[str(frame_data.data_hash)])[0]
                label_row.initialise_labels(include_signed_url=True)

            context = Context(project=project, label_row=label_row, frame_data=frame_data)
            with ExitStack() as stack:
                dependencies = solve_dependencies(context=context, dependant=dependant, stack=stack)
                func(**dependencies.values)
            return generate_response()

        return wrapper

    return context_wrapper_inner

generate_response 

generate_response() -> Response

Generate a Response object with status 200 in order to tell the FE that the function has finished successfully. :return: Response object with the right CORS settings.

Source code in encord_agents/gcp/wrappers.py 
18
19
20
21
22
23
24
25
def generate_response() -> Response:
    """
    Generate a Response object with status 200 in order to tell the FE that the function has finished successfully.
    :return: Response object with the right CORS settings.
    """
    response = make_response("")
    response.headers["Access-Control-Allow-Origin"] = "*"
    return response

FastAPI

Dependencies for injection in FastAPI servers.

This module contains dependencies that you can inject within your api routes. Dependencies that depend on others don't need to be used together. They'll work just fine alone.

Note that you can also use the function parameter: 

from typing_extensions import Annotated
from fastapi import Form
from encord_agents import FrameData
...
@app.post("/my-agent-route")
def my_agent(
    frame_data: Annotated[FrameData, Form()],
):
    ...
FrameData is automatically injected via the api request body.

encord_agents.fastapi.dependencies.dep_client 

dep_client() -> EncordUserClient

Dependency to provide an authenticated user client.

Example:

from encord.user_client import EncordUserClient
from encord_agents.fastapi.depencencies import dep_client
...
@app.post("/my-route")
def my_route(
    client: Annotated[EncordUserClient, Depends(dep_client)]
):
    # Client will authenticated and ready to use.
Source code in encord_agents/fastapi/dependencies.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def dep_client() -> EncordUserClient:
    """
    Dependency to provide an authenticated user client.

    **Example**:

    ```python
    from encord.user_client import EncordUserClient
    from encord_agents.fastapi.depencencies import dep_client
    ...
    @app.post("/my-route")
    def my_route(
        client: Annotated[EncordUserClient, Depends(dep_client)]
    ):
        # Client will authenticated and ready to use.
    ```

    """
    return get_user_client()

encord_agents.fastapi.dependencies.dep_data_lookup 

dep_data_lookup(lookup: Annotated[DataLookup, Depends(_lookup_adapter)]) -> DataLookup

Get a lookup to easily retrieve data rows and storage items associated with the given task.

Info

If you're just looking to get the associated storage item to a task, consider using dep_storage_item instead.

The lookup can, e.g., be useful for

  • Updating client metadata
  • Downloading data from signed urls
  • Matching data to other projects

Example:

from fastapi import Form
from typing_extensions import Annotated
from encord_agents import FrameData
from encord_agents.fastapi.dependencies import dep_data_lookup, DataLookup

...
@app.post("/my-agent")
def my_agent(
    frame_data: Annotated[FrameData, Form()],
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
):
    # Client will authenticated and ready to use.
    print(lookup.get_data_row(frame_data.data_hash).title)
    print(lookup.get_storage_item(frame_data.data_hash).client_metadata)

Parameters:

  • lookup (Annotated[DataLookup, Depends(_lookup_adapter)]) –

    The object that you can use to lookup data rows and storage items. Automatically injected.

Returns:

  • DataLookup –

    The (shared) lookup object.

Source code in encord_agents/fastapi/dependencies.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def dep_data_lookup(lookup: Annotated[DataLookup, Depends(_lookup_adapter)]) -> DataLookup:
    """
    Get a lookup to easily retrieve data rows and storage items associated with the given task.

    !!! info
        If you're just looking to get the associated storage item to a task, consider using `dep_storage_item` instead.


    The lookup can, e.g., be useful for

    * Updating client metadata
    * Downloading data from signed urls
    * Matching data to other projects

    **Example:**

    ```python
    from fastapi import Form
    from typing_extensions import Annotated
    from encord_agents import FrameData
    from encord_agents.fastapi.dependencies import dep_data_lookup, DataLookup

    ...
    @app.post("/my-agent")
    def my_agent(
        frame_data: Annotated[FrameData, Form()],
        lookup: Annotated[DataLookup, Depends(dep_data_lookup)]
    ):
        # Client will authenticated and ready to use.
        print(lookup.get_data_row(frame_data.data_hash).title)
        print(lookup.get_storage_item(frame_data.data_hash).client_metadata)
    ```

    Args:
        lookup: The object that you can use to lookup data rows and storage items. Automatically injected.

    Returns:
        The (shared) lookup object.

    """
    return lookup

encord_agents.fastapi.dependencies.dep_label_row 

dep_label_row(frame_data: Annotated[FrameData, Form()]) -> LabelRowV2

Dependency to provide an initialized label row.

Example:

from encord_agents.fastapi.depencencies import dep_label_row
...


@app.post("/my-route")
def my_route(
    lr: Annotated[LabelRowV2, Depends(dep_label_row)]
):
    assert lr.is_labelling_initialised  # will work

Parameters:

  • frame_data (Annotated[FrameData, Form()]) –

    the frame data from the route. This parameter is automatically injected if it's a part of your route (see example above)

Returns:

  • LabelRowV2 –

    The initialized label row.

Source code in encord_agents/fastapi/dependencies.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def dep_label_row(frame_data: Annotated[FrameData, Form()]) -> LabelRowV2:
    """
    Dependency to provide an initialized label row.

    **Example:**

    ```python
    from encord_agents.fastapi.depencencies import dep_label_row
    ...


    @app.post("/my-route")
    def my_route(
        lr: Annotated[LabelRowV2, Depends(dep_label_row)]
    ):
        assert lr.is_labelling_initialised  # will work
    ```

    Args:
        frame_data: the frame data from the route. This parameter is automatically injected
            if it's a part of your route (see example above)

    Returns:
        The initialized label row.

    """
    return get_initialised_label_row(frame_data)

encord_agents.fastapi.dependencies.dep_project 

dep_project(frame_data: Annotated[FrameData, Form()], client: Annotated[EncordUserClient, Depends(dep_client)])

Dependency to provide an instantiated Project.

Example:

from encord.project import Project
from encord_agents.fastapi.depencencies import dep_project
...
@app.post("/my-route")
def my_route(
    project: Annotated[Project, Depends(dep_project)]
):
    # Project will authenticated and ready to use.
    print(project.title)

Parameters:

  • frame_data (Annotated[FrameData, Form()]) –
  • client (Annotated[EncordUserClient, Depends(dep_client)]) –

Returns:

Source code in encord_agents/fastapi/dependencies.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def dep_project(frame_data: Annotated[FrameData, Form()], client: Annotated[EncordUserClient, Depends(dep_client)]):
    r"""
    Dependency to provide an instantiated
    [Project](https://docs.encord.com/sdk-documentation/sdk-references/LabelRowV2){ target="\_blank", rel="noopener noreferrer" }.

    **Example:**

    ```python
    from encord.project import Project
    from encord_agents.fastapi.depencencies import dep_project
    ...
    @app.post("/my-route")
    def my_route(
        project: Annotated[Project, Depends(dep_project)]
    ):
        # Project will authenticated and ready to use.
        print(project.title)
    ```


    Args:
        frame_data:
        client:

    Returns:

    """
    return client.get_project(project_hash=frame_data.project_hash)

encord_agents.fastapi.dependencies.dep_single_frame 

dep_single_frame(lr: Annotated[LabelRowV2, Depends(dep_label_row)], frame_data: Annotated[FrameData, Form()])

Dependency to inject the underlying asset of the frame data.

The downloaded asset will be named lr.data_hash.{suffix}. When the function has finished, the downloaded file will be removed from the file system.

Example:

from encord_agents.fastapi.depencencies import dep_single_frame
...

@app.post("/my-route")
def my_route(
    frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
):
    assert arr.ndim == 3, "Will work"

Parameters:

  • lr (Annotated[LabelRowV2, Depends(dep_label_row)]) –

    The label row. Automatically injected (see example above).

  • frame_data (Annotated[FrameData, Form()]) –

    the frame data from the route. This parameter is automatically injected if it's a part of your route (see example above).

Returns: Numpy array of shape [h, w, 3] RGB colors.

Source code in encord_agents/fastapi/dependencies.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def dep_single_frame(lr: Annotated[LabelRowV2, Depends(dep_label_row)], frame_data: Annotated[FrameData, Form()]):
    """
    Dependency to inject the underlying asset of the frame data.

    The downloaded asset will be named `lr.data_hash.{suffix}`.
    When the function has finished, the downloaded file will be removed from the file system.

    **Example:**

    ```python
    from encord_agents.fastapi.depencencies import dep_single_frame
    ...

    @app.post("/my-route")
    def my_route(
        frame: Annotated[NDArray[np.uint8], Depends(dep_single_frame)]
    ):
        assert arr.ndim == 3, "Will work"
    ```

    Args:
        lr: The label row. Automatically injected (see example above).
        frame_data: the frame data from the route. This parameter is automatically injected
            if it's a part of your route (see example above).

    Returns: Numpy array of shape [h, w, 3] RGB colors.

    """
    with download_asset(lr, frame_data.frame) as asset:
        img = cv2.cvtColor(cv2.imread(asset.as_posix()), cv2.COLOR_BGR2RGB)
    return np.asarray(img, dtype=np.uint8)

encord_agents.fastapi.dependencies.dep_storage_item 

dep_storage_item(lookup: Annotated[DataLookup, Depends(dep_data_lookup)], frame_data: FrameData) -> StorageItem

Get the storage item associated with the underlying agent task.

The StorageItem is useful for multiple things like

  • Updating client metadata
  • Reading file properties like storage location, fps, duration, DICOM tags, etc.

Example

from encord.storage import StorageItem
from encord_agents.fastapi.dependencies import dep_storage_item

@app.post("/my-agent")
def my_agent(
    storage_item: Annotated[StorageItem, Depends(dep_storage_item)]
):
    # Client will authenticated and ready to use.
    print(storage_item.dicom_study_uid)
    print(storage_item.client_metadata)
Source code in encord_agents/fastapi/dependencies.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def dep_storage_item(
    lookup: Annotated[DataLookup, Depends(dep_data_lookup)],
    frame_data: FrameData,
) -> StorageItem:
    r"""
    Get the storage item associated with the underlying agent task.

    The [`StorageItem`](https://docs.encord.com/sdk-documentation/sdk-references/StorageItem){ target="\_blank", rel="noopener noreferrer" }
    is useful for multiple things like

    * Updating client metadata
    * Reading file properties like storage location, fps, duration, DICOM tags, etc.

    **Example**

    ```python
    from encord.storage import StorageItem
    from encord_agents.fastapi.dependencies import dep_storage_item

    @app.post("/my-agent")
    def my_agent(
        storage_item: Annotated[StorageItem, Depends(dep_storage_item)]
    ):
        # Client will authenticated and ready to use.
        print(storage_item.dicom_study_uid)
        print(storage_item.client_metadata)
    ```

    """
    return lookup.get_storage_item(frame_data.data_hash)

encord_agents.fastapi.dependencies.dep_video_iterator 

dep_video_iterator(lr: Annotated[LabelRowV2, Depends(dep_label_row)]) -> Generator[Iterator[Frame], None, None]

Dependency to inject a video frame iterator for doing things over many frames.

Example:

from encord_agents.fastapi.depencencies import dep_video_iterator, Frame
...

@app.post("/my-route")
def my_route(
    video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
):
    for frame in video_frames:
        print(frame.frame, frame.content.shape)

Parameters:

  • lr (Annotated[LabelRowV2, Depends(dep_label_row)]) –

    Automatically injected label row dependency.

Raises:

  • NotImplementedError –

    Will fail for other data types than video.

Yields:

  • Iterator[Frame] –

    An iterator.

Source code in encord_agents/fastapi/dependencies.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def dep_video_iterator(lr: Annotated[LabelRowV2, Depends(dep_label_row)]) -> Generator[Iterator[Frame], None, None]:
    """
    Dependency to inject a video frame iterator for doing things over many frames.

    **Example:**

    ```python
    from encord_agents.fastapi.depencencies import dep_video_iterator, Frame
    ...

    @app.post("/my-route")
    def my_route(
        video_frames: Annotated[Iterator[Frame], Depends(dep_video_iterator)]
    ):
        for frame in video_frames:
            print(frame.frame, frame.content.shape)
    ```

    Args:
        lr: Automatically injected label row dependency.

    Raises:
        NotImplementedError: Will fail for other data types than video.

    Yields:
        An iterator.

    """
    if not lr.data_type == DataType.VIDEO:
        raise NotImplementedError("`dep_video_iterator` only supported for video label rows")
    with download_asset(lr, None) as asset:
        yield iter_video(asset)