Skip to content

Results Tasks

ecoscope.platform.tasks.results

Classes

DashboardJson

Bases: BaseModel

A JSON-serialized representation of a dashboard.

Attributes
filters instance-attribute 
filters: dict | None
layout instance-attribute 
layout: list
metadata instance-attribute 
metadata: Metadata
views instance-attribute 
views: dict[str, list[EmumeratedWidgetSingleView]]

OutputFiles

Bases: BaseModel

Attributes
files instance-attribute 
files: list[str | list[tuple[CompositeFilter, str]]]

SmoothingConfig

Bases: BaseModel

Configuration for data smoothing.

Attributes
degree class-attribute instance-attribute 
degree: Annotated[int, AdvancedField(default=3, description='The degree of the spline. 1: Linear, 2: Quadratic, 3: Cubic (recommended), 4-5: Higher degree.')] = 3
method class-attribute instance-attribute 
method: Annotated[Literal['spline'], Field(description="The smoothing method to apply. Currently supports 'spline'.")] = 'spline'
resolution class-attribute instance-attribute 
resolution: Annotated[int, AdvancedField(default=10, description='The resolution multiplier for interpolation points. The number of output points will be len(x) * resolution.')] = 10
y_max class-attribute instance-attribute 
y_max: Annotated[float | SkipJsonSchema[None], AdvancedField(default=None, description='The maximum value to clamp smoothed values to.')] = None
y_min class-attribute instance-attribute 
y_min: Annotated[float | SkipJsonSchema[None], AdvancedField(default=None, description='The minimum value to clamp smoothed values to. Useful for data like precipitation where values should not go below zero.')] = None

Functions:

create_geojson_layer 

create_geojson_layer(geodataframe: Annotated[AnyGeoDataFrame | SkipJsonSchema[None], Field(description='The geodataframe to visualize.', exclude=True)] = None, data_url: Annotated[str | SkipJsonSchema[None], Field(description='URL to a GeoJSON file to visualize.')] = None, layer_style: Annotated[GeoJSONLayerStyle | SkipJsonSchema[None], AdvancedField(default=GeoJSONLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description='Restrict the on-hover tooltip to these column names. None (default) shows all properties.')] = None) -> Annotated[PydeckLayerDefinition, Field()]

Creates a GeoJSON layer definition based on the provided configuration.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
@register()
def create_geojson_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame | SkipJsonSchema[None],
        Field(description="The geodataframe to visualize.", exclude=True),
    ] = None,
    data_url: Annotated[
        str | SkipJsonSchema[None],
        Field(description="URL to a GeoJSON file to visualize."),
    ] = None,
    layer_style: Annotated[
        GeoJSONLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=GeoJSONLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Restrict the on-hover tooltip to these column names. " "None (default) shows all properties."
            ),
        ),
    ] = None,
) -> Annotated[PydeckLayerDefinition, Field()]:
    """
    Creates a GeoJSON layer definition based on the provided configuration.
    """
    return PydeckLayerDefinition(
        layer_type="GeoJsonLayer",
        layer_style=layer_style or GeoJSONLayerStyle(),
        legend=legend,
        geodataframe=geodataframe,
        data_url=data_url,
        zoom=zoom,
        tooltip_columns=tooltip_columns,
    )

create_hexagon_layer 

create_hexagon_layer(geodataframe: Annotated[AnyGeoDataFrame | SkipJsonSchema[None], Field(description='The geodataframe to visualize.', exclude=True)] = None, data_url: Annotated[str | SkipJsonSchema[None], Field(description='URL to a GeoJSON file to visualize.')] = None, layer_style: Annotated[HexagonLayerStyle | SkipJsonSchema[None], AdvancedField(default=HexagonLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description='Restrict the on-hover tooltip to these column names. None (default) shows all properties.')] = None) -> Annotated[PydeckLayerDefinition, Field()]

Creates a hexagon layer definition based on the provided configuration.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
@register()
def create_hexagon_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame | SkipJsonSchema[None],
        Field(description="The geodataframe to visualize.", exclude=True),
    ] = None,
    data_url: Annotated[
        str | SkipJsonSchema[None],
        Field(description="URL to a GeoJSON file to visualize."),
    ] = None,
    layer_style: Annotated[
        HexagonLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=HexagonLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Restrict the on-hover tooltip to these column names. " "None (default) shows all properties."
            ),
        ),
    ] = None,
) -> Annotated[PydeckLayerDefinition, Field()]:
    """
    Creates a hexagon layer definition based on the provided configuration.
    """
    if geodataframe is not None:
        gdf = geodataframe.to_crs("EPSG:4326")  # type: ignore[operator]
        gdf_clean = gdf[~gdf.geometry.isna()].copy()
        gdf_clean = gdf_clean[
            ~gdf_clean.geometry.apply(lambda geom: geom.is_empty) & gdf_clean.geometry.apply(lambda geom: geom.is_valid)
        ].copy()
        # Extract longitude and latitude from geometry
        gdf_clean["lng"] = gdf_clean.geometry.x
        gdf_clean["lat"] = gdf_clean.geometry.y
        geodataframe = gdf_clean

    return PydeckLayerDefinition(
        layer_type="HexagonLayer",
        layer_style=layer_style or HexagonLayerStyle(),
        legend=legend,
        geodataframe=geodataframe,
        data_url=data_url,
        zoom=zoom,
        tooltip_columns=tooltip_columns,
    )

create_icon_layer 

create_icon_layer(geodataframe: Annotated[AnyGeoDataFrame | SkipJsonSchema[None], Field(description='The geodataframe to visualize.', exclude=True)] = None, data_url: Annotated[str | SkipJsonSchema[None], Field(description='URL to a GeoJSON file to visualize.')] = None, layer_style: Annotated[IconLayerStyle | SkipJsonSchema[None], AdvancedField(default=IconLayerStyle(), description='Style arguments for the layer.')] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description='Restrict the on-hover tooltip to these column names. None (default) shows all properties.')] = None) -> Annotated[PydeckLayerDefinition, Field()]

Creates an icon layer definition based on the provided configuration.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
@register()
def create_icon_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame | SkipJsonSchema[None],
        Field(description="The geodataframe to visualize.", exclude=True),
    ] = None,
    data_url: Annotated[
        str | SkipJsonSchema[None],
        Field(description="URL to a GeoJSON file to visualize."),
    ] = None,
    layer_style: Annotated[
        IconLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=IconLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Restrict the on-hover tooltip to these column names. " "None (default) shows all properties."
            ),
        ),
    ] = None,
) -> Annotated[PydeckLayerDefinition, Field()]:
    """
    Creates an icon layer definition based on the provided configuration.
    """
    return PydeckLayerDefinition(
        layer_type="IconLayer",
        layer_style=layer_style or IconLayerStyle(),
        # TODO support icons in legend
        legend=None,
        geodataframe=geodataframe,
        data_url=data_url,
        zoom=zoom,
        tooltip_columns=tooltip_columns,
    )

create_map_v2_widget_single_view 

create_map_v2_widget_single_view(title: Annotated[str, Field(description='The title of the widget')], data: Annotated[MapWidgetData, Field(description='A deck.gl JSON spec as a dict'), SkippedDependencyFallback(_fallback_to_none)], view: Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)] = None) -> Annotated[WidgetSingleView, Field(description='The widget')]

Create a map widget with a single view.

Parameters:

Name Type Description Default
title Annotated[str, Field(description='The title of the widget')]

The title of the widget.

 required
data Annotated[MapWidgetData, Field(description='A deck.gl JSON spec as a dict'), SkippedDependencyFallback(_fallback_to_none)]

A deck.gl JSON spec as a dict.

 required
view Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)]

If grouped, the view of the widget.

 None

Returns:

Type Description
Annotated[WidgetSingleView, Field(description='The widget')]

The widget.
Source code in ecoscope/platform/tasks/results/_widget_tasks.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@register()
def create_map_v2_widget_single_view(
    title: Annotated[str, Field(description="The title of the widget")],
    data: Annotated[
        MapWidgetData,
        Field(description="A deck.gl JSON spec as a dict"),
        SkippedDependencyFallback(_fallback_to_none),
    ],
    view: Annotated[
        CompositeFilter | None,
        Field(description="If grouped, the view of the widget", exclude=True),
    ] = None,
) -> Annotated[WidgetSingleView, Field(description="The widget")]:
    """Create a map widget with a single view.

    Args:
        title: The title of the widget.
        data: A deck.gl JSON spec as a dict.
        view: If grouped, the view of the widget.

    Returns:
        The widget.
    """
    return WidgetSingleView(
        widget_type="map_v2",
        title=title,
        view=view,
        data=data,
        is_filtered=(view is not None),
    )

create_map_widget_single_view 

create_map_widget_single_view(title: Annotated[str, Field(description='The title of the widget')], data: Annotated[PrecomputedHTMLWidgetData, Field(description='Path to precomputed HTML'), SkippedDependencyFallback(_fallback_to_none)], view: Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)] = None) -> Annotated[WidgetSingleView, Field(description='The widget')]

Create a map widget with a single view.

Parameters:

Name Type Description Default
title Annotated[str, Field(description='The title of the widget')]

The title of the widget.

 required
data Annotated[PrecomputedHTMLWidgetData, Field(description='Path to precomputed HTML'), SkippedDependencyFallback(_fallback_to_none)]

Path to precomputed HTML.

 required
view Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)]

If grouped, the view of the widget.

 None

Returns:

Type Description
Annotated[WidgetSingleView, Field(description='The widget')]

The widget.
Source code in ecoscope/platform/tasks/results/_widget_tasks.py 
26
27
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
@register()
def create_map_widget_single_view(
    title: Annotated[str, Field(description="The title of the widget")],
    data: Annotated[
        PrecomputedHTMLWidgetData,
        Field(description="Path to precomputed HTML"),
        SkippedDependencyFallback(_fallback_to_none),
    ],
    view: Annotated[
        CompositeFilter | None,
        Field(description="If grouped, the view of the widget", exclude=True),
    ] = None,
) -> Annotated[WidgetSingleView, Field(description="The widget")]:
    """Create a map widget with a single view.

    Args:
        title: The title of the widget.
        data: Path to precomputed HTML.
        view: If grouped, the view of the widget.

    Returns:
        The widget.
    """
    return WidgetSingleView(
        widget_type="map",
        title=title,
        view=view,
        data=data,
        is_filtered=(view is not None),
    )

create_path_layer 

create_path_layer(geodataframe: Annotated[AnyGeoDataFrame | SkipJsonSchema[None], Field(description='The geodataframe to visualize.', exclude=True)] = None, data_url: Annotated[str | SkipJsonSchema[None], Field(description='URL to a GeoJSON file to visualize.')] = None, layer_style: Annotated[PathLayerStyle | SkipJsonSchema[None], AdvancedField(default=PathLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description='Restrict the on-hover tooltip to these column names. None (default) shows all properties.')] = None) -> Annotated[PydeckLayerDefinition, Field()]

Creates a polyline layer definition based on the provided configuration.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
@register()
def create_path_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame | SkipJsonSchema[None],
        Field(description="The geodataframe to visualize.", exclude=True),
    ] = None,
    data_url: Annotated[
        str | SkipJsonSchema[None],
        Field(description="URL to a GeoJSON file to visualize."),
    ] = None,
    layer_style: Annotated[
        PathLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=PathLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Restrict the on-hover tooltip to these column names. " "None (default) shows all properties."
            ),
        ),
    ] = None,
) -> Annotated[PydeckLayerDefinition, Field()]:
    """
    Creates a polyline layer definition based on the provided configuration.
    """
    return PydeckLayerDefinition(
        layer_type="PathLayer",
        layer_style=layer_style or PathLayerStyle(),
        legend=legend,
        geodataframe=geodataframe,
        data_url=data_url,
        zoom=zoom,
        tooltip_columns=tooltip_columns,
    )

create_plot_widget_single_view 

create_plot_widget_single_view(title: Annotated[str, Field(description='The title of the widget')], data: Annotated[PrecomputedHTMLWidgetData, Field(description='Path to precomputed HTML'), SkippedDependencyFallback(_fallback_to_none)], view: Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)] = None) -> Annotated[WidgetSingleView, Field(description='The widget')]

Create a plot widget with a single view.

Parameters:

Name Type Description Default
title Annotated[str, Field(description='The title of the widget')]

The title of the widget.

 required
data Annotated[PrecomputedHTMLWidgetData, Field(description='Path to precomputed HTML'), SkippedDependencyFallback(_fallback_to_none)]

Path to precomputed HTML.

 required
view Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)]

If grouped, the view of the widget.

 None

Returns:

Type Description
Annotated[WidgetSingleView, Field(description='The widget')]

The widget.
Source code in ecoscope/platform/tasks/results/_widget_tasks.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@register()
def create_plot_widget_single_view(
    title: Annotated[str, Field(description="The title of the widget")],
    data: Annotated[
        PrecomputedHTMLWidgetData,
        Field(description="Path to precomputed HTML"),
        SkippedDependencyFallback(_fallback_to_none),
    ],
    view: Annotated[
        CompositeFilter | None,
        Field(description="If grouped, the view of the widget", exclude=True),
    ] = None,
) -> Annotated[WidgetSingleView, Field(description="The widget")]:
    """Create a plot widget with a single view.

    Args:
        title: The title of the widget.
        data: Path to precomputed HTML.
        view: If grouped, the view of the widget.

    Returns:
        The widget.
    """
    return WidgetSingleView(
        widget_type="graph",
        title=title,
        view=view,
        data=data,
        is_filtered=(view is not None),
    )

create_point_layer 

create_point_layer(geodataframe: Annotated[AnyGeoDataFrame, Field(description='The geodataframe to visualize.', exclude=True)], layer_style: Annotated[PointLayerStyle | SkipJsonSchema[None], AdvancedField(default=PointLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description="If present, only the listed dataframe columns will display in the layer's picking info")] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False) -> Annotated[LayerDefinition, Field()]

Creates a point layer definition based on the provided configuration.

Args: geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize. layer_style (LayerStyle): Style arguments for the data visualization.

Returns: The generated LayerDefinition

Source code in ecoscope/platform/tasks/results/_ecomap.py 
223
224
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
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
@register()
def create_point_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame,
        Field(description="The geodataframe to visualize.", exclude=True),
    ],
    layer_style: Annotated[
        PointLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=PointLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, only the listed dataframe columns will display in the layer's picking info",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
) -> Annotated[LayerDefinition, Field()]:
    """
    Creates a point layer definition based on the provided configuration.

    Args:
    geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize.
    layer_style (LayerStyle): Style arguments for the data visualization.

    Returns:
    The generated LayerDefinition
    """
    layer_style = layer_style if layer_style else PointLayerStyle()

    if isinstance(layer_style.get_radius, str):
        radius_series = geodataframe[layer_style.get_radius]

        # Lift all values up such that the min == 1
        if radius_series.min() < 0:
            radius_series = radius_series + (0 - radius_series.min()) + 1

        # set to 0, and lift everything else by 1 to distinguish NaN's and minimums
        if radius_series.hasnans:
            radius_series = radius_series + 1
            radius_series = radius_series.fillna(1)

        layer_style.get_radius = radius_series.values  # type: ignore[assignment]

    return LayerDefinition(
        geodataframe=geodataframe,
        layer_style=layer_style,
        legend=legend,  # type: ignore[arg-type]
        tooltip_columns=tooltip_columns,
        zoom=zoom,
    )

create_polygon_layer 

create_polygon_layer(geodataframe: Annotated[AnyGeoDataFrame, Field(description='The geodataframe to visualize.', exclude=True)], layer_style: Annotated[PolygonLayerStyle | SkipJsonSchema[None], AdvancedField(default=PolygonLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description="If present, only the listed dataframe columns will display in the layer's picking info")] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False) -> Annotated[LayerDefinition, Field()]

Creates a polygon layer definition based on the provided configuration.

Args: geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize. layer_style (PolygonLayerStyle): Style arguments for the data visualization.

Returns: The generated LayerDefinition

Source code in ecoscope/platform/tasks/results/_ecomap.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
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@register()
def create_polygon_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame,
        Field(description="The geodataframe to visualize.", exclude=True),
    ],
    layer_style: Annotated[
        PolygonLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=PolygonLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, only the listed dataframe columns will display in the layer's picking info",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
) -> Annotated[LayerDefinition, Field()]:
    """
    Creates a polygon layer definition based on the provided configuration.

    Args:
    geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize.
    layer_style (PolygonLayerStyle): Style arguments for the data visualization.

    Returns:
    The generated LayerDefinition
    """

    return LayerDefinition(
        geodataframe=geodataframe,
        layer_style=layer_style if layer_style else PolygonLayerStyle(),
        legend=legend,  # type: ignore[arg-type]
        tooltip_columns=tooltip_columns,
        zoom=zoom,
    )

create_polygon_layer_pydeck 

create_polygon_layer_pydeck(geodataframe: Annotated[AnyGeoDataFrame | SkipJsonSchema[None], Field(description='The geodataframe to visualize.', exclude=True)] = None, data_url: Annotated[str | SkipJsonSchema[None], Field(description='URL to a GeoJSON file to visualize.')] = None, layer_style: Annotated[PolygonLayerStyle | SkipJsonSchema[None], AdvancedField(default=PolygonLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description='Restrict the on-hover tooltip to these column names. None (default) shows all properties.')] = None) -> Annotated[PydeckLayerDefinition, Field()]

Creates a polyline layer definition based on the provided configuration.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
@register()
def create_polygon_layer_pydeck(
    geodataframe: Annotated[
        AnyGeoDataFrame | SkipJsonSchema[None],
        Field(description="The geodataframe to visualize.", exclude=True),
    ] = None,
    data_url: Annotated[
        str | SkipJsonSchema[None],
        Field(description="URL to a GeoJSON file to visualize."),
    ] = None,
    layer_style: Annotated[
        PolygonLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=PolygonLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Restrict the on-hover tooltip to these column names. " "None (default) shows all properties."
            ),
        ),
    ] = None,
) -> Annotated[PydeckLayerDefinition, Field()]:
    """
    Creates a polyline layer definition based on the provided configuration.
    """
    return PydeckLayerDefinition(
        layer_type="PolygonLayer",
        layer_style=layer_style or PolygonLayerStyle(),
        legend=legend,
        geodataframe=geodataframe,
        data_url=data_url,
        zoom=zoom,
        tooltip_columns=tooltip_columns,
    )

create_polyline_layer 

create_polyline_layer(geodataframe: Annotated[AnyGeoDataFrame, Field(description='The geodataframe to visualize.', exclude=True)], layer_style: Annotated[PolylineLayerStyle | SkipJsonSchema[None], AdvancedField(default=PolylineLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description="If present, only the listed dataframe columns will display in the layer's picking info")] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False) -> Annotated[LayerDefinition, Field()]

Creates a polyline layer definition based on the provided configuration.

Args: geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize. layer_style (PolylineLayerStyle): Style arguments for the data visualization.

Returns: The generated LayerDefinition

Source code in ecoscope/platform/tasks/results/_ecomap.py 
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@register()
def create_polyline_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame,
        Field(description="The geodataframe to visualize.", exclude=True),
    ],
    layer_style: Annotated[
        PolylineLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=PolylineLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, only the listed dataframe columns will display in the layer's picking info",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
) -> Annotated[LayerDefinition, Field()]:
    """
    Creates a polyline layer definition based on the provided configuration.

    Args:
    geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize.
    layer_style (PolylineLayerStyle): Style arguments for the data visualization.

    Returns:
    The generated LayerDefinition
    """

    return LayerDefinition(
        geodataframe=geodataframe,
        layer_style=layer_style if layer_style else PolylineLayerStyle(),
        legend=legend,
        tooltip_columns=tooltip_columns,
        zoom=zoom,
    )

create_scatterplot_layer 

create_scatterplot_layer(geodataframe: Annotated[AnyGeoDataFrame | SkipJsonSchema[None], Field(description='The geodataframe to visualize.', exclude=True)] = None, data_url: Annotated[str | SkipJsonSchema[None], Field(description='URL to a GeoJSON file to visualize.')] = None, layer_style: Annotated[ScatterplotLayerStyle | SkipJsonSchema[None], AdvancedField(default=ScatterplotLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description='Restrict the on-hover tooltip to these column names. None (default) shows all properties.')] = None) -> Annotated[PydeckLayerDefinition, Field()]

Creates a scatterplot layer definition based on the provided configuration.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
@register()
def create_scatterplot_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame | SkipJsonSchema[None],
        Field(description="The geodataframe to visualize.", exclude=True),
    ] = None,
    data_url: Annotated[
        str | SkipJsonSchema[None],
        Field(description="URL to a GeoJSON file to visualize."),
    ] = None,
    layer_style: Annotated[
        ScatterplotLayerStyle | SkipJsonSchema[None],
        AdvancedField(
            default=ScatterplotLayerStyle(),
            description="Style arguments for the layer.",
        ),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Restrict the on-hover tooltip to these column names. " "None (default) shows all properties."
            ),
        ),
    ] = None,
) -> Annotated[PydeckLayerDefinition, Field()]:
    """
    Creates a scatterplot layer definition based on the provided configuration.
    """
    layer_style = layer_style if layer_style else ScatterplotLayerStyle()
    return PydeckLayerDefinition(
        layer_type="ScatterplotLayer",
        layer_style=layer_style,
        legend=legend,
        geodataframe=geodataframe,
        data_url=data_url,
        zoom=zoom,
        tooltip_columns=tooltip_columns,
    )

create_single_value_widget_single_view 

create_single_value_widget_single_view(title: Annotated[str, Field(description='The title of the widget')], data: Annotated[Quantity | float | int | None, Field(description='Value to display.'), SkippedDependencyFallback(_fallback_to_none)], view: Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)] = None, decimal_places: Annotated[int, AdvancedField(default=1, description='The number of decimal places to display.')] = 1) -> Annotated[WidgetSingleView, Field(description='The widget')]

Create a single value widget with a single view.

Parameters:

Name Type Description Default
title Annotated[str, Field(description='The title of the widget')]

The title of the widget.

 required
data Annotated[Quantity | float | int | None, Field(description='Value to display.'), SkippedDependencyFallback(_fallback_to_none)]

The value to display.

 required
view Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)]

If grouped, the view of the widget.

 None
decimal_places Annotated[int, AdvancedField(default=1, description='The number of decimal places to display.')]

The number of decimal places to display.

 1

Returns:

Type Description
Annotated[WidgetSingleView, Field(description='The widget')]

The widget.
Source code in ecoscope/platform/tasks/results/_widget_tasks.py 
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
194
195
196
197
@register()
def create_single_value_widget_single_view(
    title: Annotated[str, Field(description="The title of the widget")],
    data: Annotated[
        Quantity | float | int | None,
        Field(description="Value to display."),
        SkippedDependencyFallback(_fallback_to_none),
    ],
    view: Annotated[
        CompositeFilter | None,
        Field(description="If grouped, the view of the widget", exclude=True),
    ] = None,
    decimal_places: Annotated[
        int,
        AdvancedField(default=1, description="The number of decimal places to display."),
    ] = 1,
) -> Annotated[WidgetSingleView, Field(description="The widget")]:
    """Create a single value widget with a single view.

    Args:
        title: The title of the widget.
        data: The value to display.
        view: If grouped, the view of the widget.
        decimal_places: The number of decimal places to display.

    Returns:
        The widget.
    """
    data_str = ""
    if data is not None:
        if isinstance(data, Quantity):
            data_str = f"{data.value:.{decimal_places}f} {data.unit or ''}".strip()
        elif isinstance(data, float):
            data_str = f"{data:.{decimal_places}f}"
        else:
            data_str = str(data)

    return WidgetSingleView(
        widget_type="stat",
        title=title,
        view=view,
        data=(data_str if data is not None else None),
        is_filtered=(view is not None),
    )

create_table_widget_single_view 

create_table_widget_single_view(title: Annotated[str, Field(description='The title of the widget')], data: Annotated[PrecomputedHTMLWidgetData, Field(description='Path to precomputed HTML'), SkippedDependencyFallback(_fallback_to_none)], view: Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)] = None) -> Annotated[WidgetSingleView, Field(description='The widget')]

Create a table widget with a single view.

Parameters:

Name Type Description Default
title Annotated[str, Field(description='The title of the widget')]

The title of the widget.

 required
data Annotated[PrecomputedHTMLWidgetData, Field(description='Path to precomputed HTML'), SkippedDependencyFallback(_fallback_to_none)]

Path to precomputed HTML.

 required
view Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)]

If grouped, the view of the widget.

 None

Returns:

Type Description
Annotated[WidgetSingleView, Field(description='The widget')]

The widget.
Source code in ecoscope/platform/tasks/results/_widget_tasks.py 
200
201
202
203
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
@register()
def create_table_widget_single_view(
    title: Annotated[str, Field(description="The title of the widget")],
    data: Annotated[
        PrecomputedHTMLWidgetData,
        Field(description="Path to precomputed HTML"),
        SkippedDependencyFallback(_fallback_to_none),
    ],
    view: Annotated[
        CompositeFilter | None,
        Field(description="If grouped, the view of the widget", exclude=True),
    ] = None,
) -> Annotated[WidgetSingleView, Field(description="The widget")]:
    """Create a table widget with a single view.

    Args:
        title: The title of the widget.
        data: Path to precomputed HTML.
        view: If grouped, the view of the widget.

    Returns:
        The widget.
    """
    return WidgetSingleView(
        widget_type="table",
        title=title,
        view=view,
        data=data,
        is_filtered=(view is not None),
    )

create_text_layer 

create_text_layer(geodataframe: Annotated[AnyGeoDataFrame, Field(description='The geodataframe to visualize.', exclude=True)], layer_style: Annotated[TextLayerStyle | SkipJsonSchema[None], AdvancedField(default=TextLayerStyle(), description='Style arguments for the layer.')] = None) -> Annotated[LayerDefinition, Field()]

Creates a text layer definition based on the provided configuration.

Args: geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize. layer_style (LayerStyle): Style arguments for the data visualization.

Returns: The generated LayerDefinition

Source code in ecoscope/platform/tasks/results/_ecomap.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
@register()
def create_text_layer(
    geodataframe: Annotated[
        AnyGeoDataFrame,
        Field(description="The geodataframe to visualize.", exclude=True),
    ],
    layer_style: Annotated[
        TextLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=TextLayerStyle(), description="Style arguments for the layer."),
    ] = None,
) -> Annotated[LayerDefinition, Field()]:
    """
    Creates a text layer definition based on the provided configuration.

    Args:
    geodataframe (geopandas.GeoDataFrame): The geodataframe to visualize.
    layer_style (LayerStyle): Style arguments for the data visualization.

    Returns:
    The generated LayerDefinition
    """

    return LayerDefinition(
        geodataframe=geodataframe,
        layer_style=layer_style if layer_style else TextLayerStyle(),
        legend=None,
        tooltip_columns=None,
    )

create_text_layer_pydeck 

create_text_layer_pydeck(geodataframe: Annotated[AnyGeoDataFrame | SkipJsonSchema[None], Field(description='The geodataframe to visualize.', exclude=True)] = None, data_url: Annotated[str | SkipJsonSchema[None], Field(description='URL to a GeoJSON file to visualize.')] = None, layer_style: Annotated[TextLayerStyle | SkipJsonSchema[None], AdvancedField(default=TextLayerStyle(), description='Style arguments for the layer.')] = None, legend: Annotated[LegendDefinition | SkipJsonSchema[None], AdvancedField(default=None, description='If present, includes this layer in the map legend')] = None, zoom: Annotated[bool, AdvancedField(default=False, description='If true, the map will be zoomed to the bounds of this layer', exclude=True)] = False, tooltip_columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(default=None, description='Restrict the on-hover tooltip to these column names. None (default) shows all properties.')] = None) -> Annotated[PydeckLayerDefinition, Field()]

Creates a text layer definition based on the provided configuration.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
@register()
def create_text_layer_pydeck(
    geodataframe: Annotated[
        AnyGeoDataFrame | SkipJsonSchema[None],
        Field(description="The geodataframe to visualize.", exclude=True),
    ] = None,
    data_url: Annotated[
        str | SkipJsonSchema[None],
        Field(description="URL to a GeoJSON file to visualize."),
    ] = None,
    layer_style: Annotated[
        TextLayerStyle | SkipJsonSchema[None],
        AdvancedField(default=TextLayerStyle(), description="Style arguments for the layer."),
    ] = None,
    legend: Annotated[
        LegendDefinition | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="If present, includes this layer in the map legend",
        ),
    ] = None,
    zoom: Annotated[
        bool,
        AdvancedField(
            default=False,
            description="If true, the map will be zoomed to the bounds of this layer",
            exclude=True,
        ),
    ] = False,
    tooltip_columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Restrict the on-hover tooltip to these column names. " "None (default) shows all properties."
            ),
        ),
    ] = None,
) -> Annotated[PydeckLayerDefinition, Field()]:
    """
    Creates a text layer definition based on the provided configuration.
    """
    return PydeckLayerDefinition(
        layer_type="TextLayer",
        layer_style=layer_style or TextLayerStyle(),
        legend=legend,
        geodataframe=geodataframe,
        data_url=data_url,
        zoom=zoom,
        tooltip_columns=tooltip_columns,
    )

create_text_widget_single_view 

create_text_widget_single_view(title: Annotated[str, Field(description='The title of the widget')], data: Annotated[TextWidgetData, Field(description='Text to display.'), SkippedDependencyFallback(_fallback_to_none)], view: Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)] = None) -> Annotated[WidgetSingleView, Field(description='The widget')]

Create a text widget with a single view.

Parameters:

Name Type Description Default
title Annotated[str, Field(description='The title of the widget')]

The title of the widget.

 required
data Annotated[TextWidgetData, Field(description='Text to display.'), SkippedDependencyFallback(_fallback_to_none)]

Text to display.

 required
view Annotated[CompositeFilter | None, Field(description='If grouped, the view of the widget', exclude=True)]

If grouped, the view of the widget.

 None

Returns:

Type Description
Annotated[WidgetSingleView, Field(description='The widget')]

The widget.
Source code in ecoscope/platform/tasks/results/_widget_tasks.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@register()
def create_text_widget_single_view(
    title: Annotated[str, Field(description="The title of the widget")],
    data: Annotated[
        TextWidgetData,
        Field(description="Text to display."),
        SkippedDependencyFallback(_fallback_to_none),
    ],
    view: Annotated[
        CompositeFilter | None,
        Field(description="If grouped, the view of the widget", exclude=True),
    ] = None,
) -> Annotated[WidgetSingleView, Field(description="The widget")]:
    """Create a text widget with a single view.

    Args:
        title: The title of the widget.
        data: Text to display.
        view: If grouped, the view of the widget.

    Returns:
        The widget.
    """
    return WidgetSingleView(
        widget_type="text",
        title=title,
        view=view,
        data=data,
        is_filtered=(view is not None),
    )

create_tiled_bitmap_layer 

create_tiled_bitmap_layer(url: Annotated[str, Field(description='The tile URL template with {z}, {x}, {y} placeholders.')], opacity: Annotated[float, Field(description='Layer opacity from 0 to 1.', ge=0, le=1)] = 1.0, max_zoom: Annotated[int, AdvancedField(description='Maximum zoom level.', default=20)] = 20, min_zoom: Annotated[int, AdvancedField(description='Minimum zoom level.', default=0)] = 0) -> Annotated[TileLayer, Field()]

Creates a tiled bitmap layer definition from a tile URL.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
@register()
def create_tiled_bitmap_layer(
    url: Annotated[
        str,
        Field(
            description="The tile URL template with {z}, {x}, {y} placeholders.",
        ),
    ],
    opacity: Annotated[
        float,
        Field(
            description="Layer opacity from 0 to 1.",
            ge=0,
            le=1,
        ),
    ] = 1.0,
    max_zoom: Annotated[
        int,
        AdvancedField(
            description="Maximum zoom level.",
            default=20,
        ),
    ] = 20,
    min_zoom: Annotated[
        int,
        AdvancedField(
            description="Minimum zoom level.",
            default=0,
        ),
    ] = 0,
) -> Annotated[TileLayer, Field()]:
    """Creates a tiled bitmap layer definition from a tile URL."""
    return TileLayer(
        url=url,
        opacity=opacity,
        max_zoom=max_zoom,
        min_zoom=min_zoom,
    )

draw_bar_chart 

draw_bar_chart(dataframe: DataFrame[JsonSerializableDataFrameModel], bar_chart_configs: Annotated[list[BarConfig], Field(description='Bar chart configuration.', title='Bar Chart Configuration')], category: Annotated[str, Field(description='The column name in the dataframe to group by and use as the x-axis categories.')], layout_kwargs: Annotated[LayoutStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs passed to plotly.go.Figure(layout).')] = None, widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Generates a bar chart from the provided params

Args: dataframe (pd.DataFrame): The input dataframe. bar_configs (list[BarConfig]): a list of BarConfigs specifying the the bar chart labels, columns, and functions for aggregation. category (str): The column name in the dataframe to group by and use as the x-axis categories. layout_kwargs (LayoutStyle): Additional styling options passed to plotly.go.Figure(layout).

Returns: The generated chart html as a string

Source code in ecoscope/platform/tasks/results/_ecoplot.py 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
@register()
def draw_bar_chart(
    dataframe: DataFrame[JsonSerializableDataFrameModel],
    bar_chart_configs: Annotated[
        list[BarConfig],
        Field(description="Bar chart configuration.", title="Bar Chart Configuration"),
    ],
    category: Annotated[
        str,
        Field(description="The column name in the dataframe to group by and use as the x-axis categories."),
    ],
    layout_kwargs: Annotated[
        LayoutStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs passed to plotly.go.Figure(layout).",
        ),
    ] = None,
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Generates a bar chart from the provided params

    Args:
    dataframe (pd.DataFrame): The input dataframe.
    bar_configs (list[BarConfig]): a list of BarConfigs
        specifying the the bar chart labels, columns, and functions for aggregation.
    category (str): The column name in the dataframe to group by and use as the x-axis categories.
    layout_kwargs (LayoutStyle): Additional styling options passed to plotly.go.Figure(layout).

    Returns:
    The generated chart html as a string
    """
    import ecoscope.plotting as ecoplot

    bar_configs = [
        ecoplot.BarConfig(
            column=config.column,
            agg_func=config.agg_func,
            label=config.label,
            show_label=config.show_label,
            style=config.style.model_dump(exclude_none=True) if config.style else {},
        )
        for config in bar_chart_configs
    ]
    plot = ecoplot.bar_chart(
        data=dataframe,
        bar_configs=bar_configs,
        category=category,
        layout_kwargs=layout_kwargs.model_dump(exclude_none=True) if layout_kwargs else {},
    )

    return plot.to_html(**ExportArgs(div_id=widget_id).model_dump(exclude_none=True))

draw_ecomap 

draw_ecomap(geo_layers: Annotated[LayerDefinition | list[LayerDefinition], Field(description='A list of map layers to add to the map.', exclude=True)], tile_layers: Annotated[list[TileLayer] | SkipJsonSchema[None], Field(description='A list of named tile layer with opacity, ie OpenStreetMap.')] = None, static: Annotated[bool, Field(description='Set to true to disable map pan/zoom.')] = False, title: Annotated[str | SkipJsonSchema[None], AdvancedField(default=None, description='            The map title. Note this is the title drawn on the map canvas itself, and will result\n            in duplicate titles if set in the context of a dashboard in which the iframe/widget\n            container also has a title set on it.\n            ')] = None, north_arrow_style: Annotated[NorthArrowStyle | SkipJsonSchema[None], Field(description='Additional arguments for configuring the North Arrow.')] = None, legend_style: Annotated[LegendStyle | SkipJsonSchema[None], Field(description='Additional arguments for configuring the legend.')] = None, max_zoom: Annotated[int, Field(description='Max zoom level.')] = 20, view_state: Annotated[ViewState | SkipJsonSchema[None], Field(description='Manually set the view state of the map, overrides any layer zoom settings.', exclude=True)] = None, widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Creates a map based on the provided layer definitions and configuration.

Args: geo_layers (LayerDefinition | list[LayerDefinition]): A list of map layers to add to the map. tile_layers (list): A named tile layer, ie OpenStreetMap. static (bool): Set to true to disable map pan/zoom. title (str): The map title. north_arrow_style (NorthArrowStyle): Additional arguments for configuring the North Arrow. legend_style (WidgetStyleBase): Additional arguments for configuring the Legend. max_zoom (int): The maximum zoom level of the map view_state (ViewState): Manually set the view state of the map, overrides any layer zoom settings. widget_id (str): The id of the dashboard widget that this tile layer belongs to. If set this MUST match the widget title as defined downstream in create_widget tasks

Returns: str: A static HTML representation of the map.

Source code in ecoscope/platform/tasks/results/_ecomap.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
@register()
def draw_ecomap(
    geo_layers: Annotated[
        LayerDefinition | list[LayerDefinition],
        Field(description="A list of map layers to add to the map.", exclude=True),
    ],
    tile_layers: Annotated[
        list[TileLayer] | SkipJsonSchema[None],
        Field(description="A list of named tile layer with opacity, ie OpenStreetMap."),
    ] = None,
    static: Annotated[bool, Field(description="Set to true to disable map pan/zoom.")] = False,
    title: Annotated[
        str | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="""\
            The map title. Note this is the title drawn on the map canvas itself, and will result
            in duplicate titles if set in the context of a dashboard in which the iframe/widget
            container also has a title set on it.
            """,
        ),
    ] = None,
    north_arrow_style: Annotated[
        NorthArrowStyle | SkipJsonSchema[None],
        Field(description="Additional arguments for configuring the North Arrow."),
    ] = None,
    legend_style: Annotated[
        LegendStyle | SkipJsonSchema[None],
        Field(description="Additional arguments for configuring the legend."),
    ] = None,
    max_zoom: Annotated[
        int,
        Field(description="Max zoom level."),
    ] = 20,
    view_state: Annotated[
        ViewState | SkipJsonSchema[None],
        Field(
            description="Manually set the view state of the map, overrides any layer zoom settings.",
            exclude=True,
        ),
    ] = None,
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Creates a map based on the provided layer definitions and configuration.

    Args:
    geo_layers (LayerDefinition | list[LayerDefinition]): A list of map layers to add to the map.
    tile_layers (list): A named tile layer, ie OpenStreetMap.
    static (bool): Set to true to disable map pan/zoom.
    title (str): The map title.
    north_arrow_style (NorthArrowStyle): Additional arguments for configuring the North Arrow.
    legend_style (WidgetStyleBase): Additional arguments for configuring the Legend.
    max_zoom (int): The maximum zoom level of the map
    view_state (ViewState): Manually set the view state of the map, overrides any layer zoom settings.
    widget_id (str): The id of the dashboard widget that this tile layer belongs to.
        If set this MUST match the widget title as defined downstream in create_widget tasks

    Returns:
    str: A static HTML representation of the map.
    """
    import pandas as pd  # type: ignore[import-untyped]
    from lonboard import BitmapTileLayer  # type: ignore[import-untyped]
    from lonboard.experimental import TextLayer  # type: ignore[import-untyped]

    from ecoscope.mapping import EcoMap

    if tile_layers is None:
        tile_layers = []
    if north_arrow_style is None:
        north_arrow_style = NorthArrowStyle()
    if legend_style is None:
        legend_style = LegendStyle()

    legend_labels: list = []
    legend_colors: list = []
    zoom_layers: list = []

    m = EcoMap(static=static, default_widgets=False)

    if title:
        m.add_title(title)

    m.add_scale_bar()
    m.add_north_arrow(**(north_arrow_style.model_dump(exclude_none=True)))  # type: ignore[union-attr]
    m.add_save_image()

    for tile_layer in tile_layers:
        layer = BitmapTileLayer(
            data=tile_layer.url,  # type: ignore[arg-type]
            max_zoom=tile_layer.max_zoom,  # type: ignore[arg-type]
            min_zoom=tile_layer.min_zoom,  # type: ignore[arg-type]
            opacity=tile_layer.opacity,  # type: ignore[arg-type]
            tile_size=256,  # type: ignore[arg-type]
            widget_id=widget_id,  # type: ignore[arg-type]
        )
        m.add_layer(layer)

    geo_layers = [geo_layers] if not isinstance(geo_layers, list) else geo_layers
    for layer_def in geo_layers:
        match layer_def.layer_style:
            case PointLayerStyle():
                layer = EcoMap.point_layer(
                    layer_def.geodataframe,
                    tooltip_columns=layer_def.tooltip_columns,
                    **layer_def.layer_style.model_dump(exclude_none=True),
                )
            case PolylineLayerStyle():
                layer = EcoMap.polyline_layer(
                    layer_def.geodataframe,
                    tooltip_columns=layer_def.tooltip_columns,
                    **layer_def.layer_style.model_dump(exclude_none=True),
                )
            case PolygonLayerStyle():
                layer = EcoMap.polygon_layer(
                    layer_def.geodataframe,
                    tooltip_columns=layer_def.tooltip_columns,
                    **layer_def.layer_style.model_dump(exclude_none=True),
                )
            case TextLayerStyle():
                ls = layer_def.layer_style
                layer = TextLayer.from_geopandas(
                    layer_def.geodataframe,  # type: ignore[call-arg]
                    get_text=layer_def.geodataframe.label,  # type: ignore[call-arg]
                    get_color=ls.get_color,  # type: ignore[call-arg]
                    font_family=ls.font_family,  # type: ignore[call-arg]
                    font_weight=ls.font_weight,  # type: ignore[call-arg]
                    get_size=ls.get_size,  # type: ignore[call-arg]
                    get_text_anchor=ls.get_text_anchor,  # type: ignore[call-arg]
                    get_alignment_baseline=ls.get_alignment_baseline,  # type: ignore[call-arg]
                    get_background_color=ls.get_background_color,  # type: ignore[call-arg]
                    pickable=ls.pickable,  # type: ignore[call-arg]
                )

        if layer_def.legend:
            if layer_def.legend.label_column and layer_def.legend.color_column:
                lookup = layer_def.geodataframe.drop_duplicates(subset=layer_def.legend.label_column)
                if layer_def.legend.sort:
                    lookup = lookup.sort_values(
                        layer_def.legend.label_column,
                        ascending=True if layer_def.legend.sort == "ascending" else False,
                        # Attempt to coerce numeric strings to numbers
                        # in order to maintain a proper numeric sort
                        key=lambda col: pd.to_numeric(col, errors="ignore"),  # type: ignore[call-overload]
                    )
                for _, row in lookup.iterrows():
                    legend_labels.append(row[layer_def.legend.label_column])
                    legend_colors.append(row[layer_def.legend.color_column])
            elif layer_def.legend.labels and layer_def.legend.colors:
                legend_labels.extend(layer_def.legend.labels)
                legend_colors.extend(layer_def.legend.colors)
            if legend_labels and layer_def.legend.label_suffix:
                legend_labels = [label + layer_def.legend.label_suffix for label in legend_labels]

        if layer_def.zoom:
            zoom_layers.append(layer)

        m.add_layer(layer)

    if len(legend_labels) > 0:
        m.add_legend(
            labels=[str(ll) for ll in legend_labels],
            colors=legend_colors,
            title=legend_style.display_name,
            placement=legend_style.placement,
        )

    if view_state is None:
        if not zoom_layers:
            zoom_layers = m.layers
        m.zoom_to_bounds(feat=zoom_layers, max_zoom=max_zoom)
    else:
        m.set_view_state(**view_state.model_dump())

    return m.to_html()

draw_ecoplot 

draw_ecoplot(dataframe: DataFrame[JsonSerializableDataFrameModel], group_by: Annotated[str, Field(description='The dataframe column to group by.')], ecoplot_configs: Annotated[list[EcoplotConfig], Field(description='ecoplot configs.')], tickformat: Annotated[str, AdvancedField(default='%b-%Y', description='The time format for timeseries data.')] = '%b-%Y', widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Generates an EcoPlot from the provided params

Args: dataframe (pd.DataFrame): The input dataframe. group_by (str): The dataframe column to group by. ecoplot_configs (list[EcoplotConfig]): the ecoplot configs widget_id (str): The id of the dashboard widget that this tile layer belongs to. If set this MUST match the widget title as defined downstream in create_widget tasks

Returns: The generated plot html as a string

Source code in ecoscope/platform/tasks/results/_ecoplot.py 
184
185
186
187
188
189
190
191
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
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@register()
def draw_ecoplot(
    dataframe: DataFrame[JsonSerializableDataFrameModel],
    group_by: Annotated[str, Field(description="The dataframe column to group by.")],
    ecoplot_configs: Annotated[list[EcoplotConfig], Field(description="ecoplot configs.")],
    tickformat: Annotated[
        str,
        AdvancedField(default="%b-%Y", description="The time format for timeseries data."),
    ] = "%b-%Y",
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Generates an EcoPlot from the provided params

    Args:
    dataframe (pd.DataFrame): The input dataframe.
    group_by (str): The dataframe column to group by.
    ecoplot_configs (list[EcoplotConfig]): the ecoplot configs
    widget_id (str): The id of the dashboard widget that this tile layer belongs to.
        If set this MUST match the widget title as defined downstream in create_widget tasks

    Returns:
    The generated plot html as a string
    """
    import ecoscope.plotting as plotting

    grouped = dataframe.groupby(group_by)

    data = []
    for config in ecoplot_configs:
        data.append(
            plotting.EcoPlotData(
                grouped=grouped,
                x_col=config.x_col,
                y_col=config.y_col,
                color_col=config.color_col,
                **(config.plot_style.model_dump(exclude_none=True) if config.plot_style else {}),
            )
        )

    plot = plotting.ecoplot(
        data=data,
        tickformat=tickformat,
    )

    return plot.to_html(**ExportArgs(div_id=widget_id).model_dump(exclude_none=True))

draw_historic_timeseries 

draw_historic_timeseries(dataframe: DataFrame[JsonSerializableDataFrameModel], current_value_column: Annotated[str, Field(description='The name of the dataframe column to pull slice values from')], current_value_title: Annotated[str, Field(description='The title shown in the plot legend for current value')], historic_min_column: Annotated[str | SkipJsonSchema[None], Field(description='The name of the dataframe column to pull historic min values from')] = None, historic_max_column: Annotated[str | SkipJsonSchema[None], Field(description='The name of the dataframe column to pull historic max values from')] = None, historic_band_title: Annotated[str | SkipJsonSchema[None], Field(description='The title shown in the plot legend for historic band')] = 'Historic Min-Max', historic_mean_column: Annotated[str | SkipJsonSchema[None], Field(description='The name of the dataframe column to pull historic mean values from')] = None, historic_mean_title: Annotated[str | SkipJsonSchema[None], Field(description='The title shown in the plot legend for historic mean values')] = 'Historic Mean', layout_style: Annotated[LayoutStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs passed to plotly.go.Figure(layout).')] = None, upper_lower_band_style: Annotated[PlotStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs for upper_lower_band passed to plotly.graph_objects.Scatter.')] = PlotStyle(mode='lines', line=LineStyle(color='green')), historic_mean_style: Annotated[PlotStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs passed for historic_mean to plotly.graph_objects.Scatter.')] = None, current_value_style: Annotated[PlotStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs for current_value passed to plotly.graph_objects.Scatter.')] = None, time_column: Annotated[str | SkipJsonSchema[None], Field(description='The name of the dataframe column to pull historic max values from')] = 'img_date', widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Creates a timeseries plot compared with historical values Parameters

df: pd.Dataframe The data to plot current_value_column: str The name of the dataframe column to pull slice values from current_value_title: str The title of the current value historic_min_column: str The name of the dataframe column to pull historic min values from. historic_min_column and historic_max_column should exist together. historic_max_column: str The name of the dataframe column to pull historic max values from. historic_min_column and historic_max_column should exist together. historic_mean_column: str The name of the dataframe column to pull historic mean values from layout_kwargs: dict Additional kwargs passed to plotly.go.Figure(layout) upper_lower_band_style: PlotStyle Additional kwargs for upper_lower_band passed to plotly.graph_objects.Scatter historic_mean_style: PlotStyle Additional kwargs passed for historic_mean to plotly.graph_objects.Scatter current_value_style: PlotStyle Additional kwargs for current_value passed to plotly.graph_objects.Scatter time_column: str The name of the dataframe column to pull time values from widget_id str: The id of the dashboard widget that this tile layer belongs to. If set this MUST match the widget title as defined downstream in create_widget tasks Returns

fig : The generated chart html as a string

Source code in ecoscope/platform/tasks/results/_ecoplot.py 
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
@register()
def draw_historic_timeseries(
    dataframe: DataFrame[JsonSerializableDataFrameModel],
    current_value_column: Annotated[
        str,
        Field(description="The name of the dataframe column to pull slice values from"),
    ],
    current_value_title: Annotated[
        str,
        Field(description="The title shown in the plot legend for current value"),
    ],
    historic_min_column: Annotated[
        str | SkipJsonSchema[None],
        Field(description="The name of the dataframe column to pull historic min values from"),
    ] = None,
    historic_max_column: Annotated[
        str | SkipJsonSchema[None],
        Field(description="The name of the dataframe column to pull historic max values from"),
    ] = None,
    historic_band_title: Annotated[
        str | SkipJsonSchema[None],
        Field(description="The title shown in the plot legend for historic band"),
    ] = "Historic Min-Max",
    historic_mean_column: Annotated[
        str | SkipJsonSchema[None],
        Field(description="The name of the dataframe column to pull historic mean values from"),
    ] = None,
    historic_mean_title: Annotated[
        str | SkipJsonSchema[None],
        Field(description="The title shown in the plot legend for historic mean values"),
    ] = "Historic Mean",
    layout_style: Annotated[
        LayoutStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs passed to plotly.go.Figure(layout).",
        ),
    ] = None,
    upper_lower_band_style: Annotated[
        PlotStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs for upper_lower_band passed to plotly.graph_objects.Scatter.",
        ),
    ] = PlotStyle(mode="lines", line=LineStyle(color="green")),
    historic_mean_style: Annotated[
        PlotStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs passed for historic_mean to plotly.graph_objects.Scatter.",
        ),
    ] = None,
    current_value_style: Annotated[
        PlotStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs for current_value passed to plotly.graph_objects.Scatter.",
        ),
    ] = None,
    time_column: Annotated[
        str | SkipJsonSchema[None],
        Field(description="The name of the dataframe column to pull historic max values from"),
    ] = "img_date",
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Creates a timeseries plot compared with historical values
    Parameters
    ----------
    df: pd.Dataframe
        The data to plot
    current_value_column: str
        The name of the dataframe column to pull slice values from
    current_value_title: str
        The title of the current value
    historic_min_column: str
        The name of the dataframe column to pull historic min values from.
        historic_min_column and historic_max_column should exist together.
    historic_max_column: str
        The name of the dataframe column to pull historic max values from.
        historic_min_column and historic_max_column should exist together.
    historic_mean_column: str
        The name of the dataframe column to pull historic mean values from
    layout_kwargs: dict
        Additional kwargs passed to plotly.go.Figure(layout)
    upper_lower_band_style: PlotStyle
        Additional kwargs for upper_lower_band passed to plotly.graph_objects.Scatter
    historic_mean_style: PlotStyle
        Additional kwargs passed for historic_mean to plotly.graph_objects.Scatter
    current_value_style: PlotStyle
        Additional kwargs for current_value passed to plotly.graph_objects.Scatter
    time_column: str
        The name of the dataframe column to pull time values from
    widget_id str: The id of the dashboard widget that this tile layer belongs to.
        If set this MUST match the widget title as defined downstream in create_widget tasks
    Returns
    -------
    fig : The generated chart html as a string
    """
    from ecoscope.plotting.plot import draw_historic_timeseries

    if historic_mean_style is None:
        historic_mean_style = PlotStyle(mode="lines", line=LineStyle(color="green", dash="dot"))
    if current_value_style is None:
        current_value_style = PlotStyle(mode="lines", line=LineStyle(color="navy"))

    fig = draw_historic_timeseries(
        dataframe,
        current_value_column=current_value_column,
        current_value_title=current_value_title,
        time_column=time_column,  # type: ignore[arg-type]
        historic_min_column=historic_min_column,
        historic_max_column=historic_max_column,
        historic_band_title=historic_band_title,  # type: ignore[arg-type]
        historic_mean_column=historic_mean_column,
        historic_mean_title=historic_mean_title,  # type: ignore[arg-type]
        layout_kwargs=layout_style.model_dump(exclude_none=True) if layout_style else {},
        upper_lower_band_style=upper_lower_band_style.model_dump(exclude_none=True) if upper_lower_band_style else {},
        historic_mean_style=historic_mean_style.model_dump(exclude_none=True) if historic_mean_style else {},
        current_value_style=current_value_style.model_dump(exclude_none=True) if current_value_style else {},
    )

    return fig.to_html(**ExportArgs(div_id=widget_id).model_dump(exclude_none=True))

draw_line_chart 

draw_line_chart(dataframe: DataFrame[JsonSerializableDataFrameModel], x_column: Annotated[str, Field(description='The dataframe column to plot in the x/time axis.')], y_column: Annotated[str, Field(description='The dataframe column to plot in the y/time axis.')], category_column: Annotated[str | SkipJsonSchema[None], Field(description='The column name in the dataframe to group by and plot separate traces.')] = None, line_kwargs: Annotated[LineStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Line style settings')] = None, layout_kwargs: Annotated[LayoutStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs passed to plotly.go.Figure(layout).')] = None, smoothing: Annotated[SmoothingConfig | SkipJsonSchema[None], AdvancedField(default=None, description='Configuration for line smoothing. When set, creates a smoothed line with original data point markers.')] = None, widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Generates a line chart from the provided params

Args: dataframe (pd.DataFrame): The input dataframe. x_column (str): The dataframe column to plot in the x/time axis. y_column (str): The dataframe column to plot in the y/time axis. category_column (str): The column name in the dataframe to group by and plot separate traces. line_kwargs (LineStyle): Additional styling options passed to each line of the chart. layout_kwargs (LayoutStyle): Additional styling styling options passed to plotly.go.Figure(layout). smoothing (SmoothingConfig): Configuration for line smoothing. When set, creates two layers: a smoothed line and original data point markers.

Returns: The generated chart html as a string

Source code in ecoscope/platform/tasks/results/_ecoplot.py 
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
@register()
def draw_line_chart(
    dataframe: DataFrame[JsonSerializableDataFrameModel],
    x_column: Annotated[str, Field(description="The dataframe column to plot in the x/time axis.")],
    y_column: Annotated[str, Field(description="The dataframe column to plot in the y/time axis.")],
    category_column: Annotated[
        str | SkipJsonSchema[None],
        Field(description="The column name in the dataframe to group by and plot separate traces."),
    ] = None,
    line_kwargs: Annotated[
        LineStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Line style settings",
        ),
    ] = None,
    layout_kwargs: Annotated[
        LayoutStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs passed to plotly.go.Figure(layout).",
        ),
    ] = None,
    smoothing: Annotated[
        SmoothingConfig | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "Configuration for line smoothing. When set, creates a smoothed line with original data point markers."
            ),
        ),
    ] = None,
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Generates a line chart from the provided params

    Args:
    dataframe (pd.DataFrame): The input dataframe.
    x_column (str): The dataframe column to plot in the x/time axis.
    y_column (str): The dataframe column to plot in the y/time axis.
    category_column (str): The column name in the dataframe to group by and plot separate traces.
    line_kwargs (LineStyle): Additional styling options passed to each line of the chart.
    layout_kwargs (LayoutStyle): Additional styling styling options passed to plotly.go.Figure(layout).
    smoothing (SmoothingConfig): Configuration for line smoothing. When set, creates two layers:
        a smoothed line and original data point markers.

    Returns:
    The generated chart html as a string
    """
    import ecoscope.plotting as ecoplot
    from ecoscope.analysis.smoothing import (
        SmoothingConfig as EcoSmoothingConfig,
    )

    smoothing_config = None
    if smoothing is not None:
        smoothing_config = EcoSmoothingConfig(
            method=smoothing.method,
            y_min=smoothing.y_min,
            y_max=smoothing.y_max,
            resolution=smoothing.resolution,
            degree=smoothing.degree,
        )

    plot = ecoplot.line_chart(
        data=dataframe,
        x_column=x_column,
        y_column=y_column,
        category_column=category_column,
        line_kwargs=line_kwargs.model_dump(exclude_none=True) if line_kwargs else {},
        layout_kwargs=layout_kwargs.model_dump(exclude_none=True) if layout_kwargs else {},
        smoothing=smoothing_config,
    )

    return plot.to_html(**ExportArgs(div_id=widget_id).model_dump(exclude_none=True))

draw_map 

draw_map(geo_layers: Annotated[PydeckLayerDefinition | list[PydeckLayerDefinition] | SkipJsonSchema[None], Field(description='A list of map layers to add to the map.', exclude=True)] = None, tile_layers: Annotated[list[TileLayer | BitmapLayerDefinition] | SkipJsonSchema[None], Field(description='A list of tile layers (base maps and/or overlays).')] = None, static: Annotated[bool, Field(description='Set to true to disable map pan/zoom.')] = False, output_type: Annotated[Literal['html', 'json'], Field(description='Whether to return rendered HTML or a deck.gl JSON spec dict.')] = 'html', title: Annotated[str | SkipJsonSchema[None], AdvancedField(default='', description='            The map title. Note this is the title drawn on the map canvas itself, and will result\n            in duplicate titles if set in the context of a dashboard in which the iframe/widget\n            container also has a title set on it.\n            ')] = None, legend_style: Annotated[LegendStyle | SkipJsonSchema[None], AdvancedField(default=LegendStyle(), description='Additional arguments for configuring the legend.')] = None, max_zoom: Annotated[int, AdvancedField(default=20, description='            The maximum zoom level allowed by the map.\n            This setting will be overridden if provided\n            tile layers max zoom levels are lower than this value.\n            ')] = 20, view_state: Annotated[ViewState | SkipJsonSchema[None], AdvancedField(default=ViewState(), description='Manually set the view state of the map.')] = None, widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str | DeckJsonSpec, Field()]

Creates a map based on the provided layer definitions and configuration.

Args: geo_layers (PydeckLayerDefinition | list[PydeckLayerDefinition] | None): Map layers to add to the map. tile_layers (list): A named tile layer, ie OpenStreetMap. static (bool): Set to true to disable map pan/zoom. title (str): The map title. legend_style (WidgetStyleBase): Additional arguments for configuring the Legend. max_zoom (int): The maximum zoom level of the map view_state (ViewState): Manually set the view state of the map, overrides any layer zoom settings. widget_id (str): The id of the dashboard widget that this tile layer belongs to. If set this MUST match the widget title as defined downstream in create_widget tasks

str | DeckJsonSpec: A static HTML representation of the map, or a validated deck.gl JSON spec if output_type="json".

Source code in ecoscope/platform/tasks/results/_pydeck.py 
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
@register()
def draw_map(
    geo_layers: Annotated[
        PydeckLayerDefinition | list[PydeckLayerDefinition] | SkipJsonSchema[None],
        Field(description="A list of map layers to add to the map.", exclude=True),
    ] = None,
    tile_layers: Annotated[
        list[TileLayer | BitmapLayerDefinition] | SkipJsonSchema[None],
        Field(description="A list of tile layers (base maps and/or overlays)."),
    ] = None,
    static: Annotated[bool, Field(description="Set to true to disable map pan/zoom.")] = False,
    output_type: Annotated[
        Literal["html", "json"],
        Field(description="Whether to return rendered HTML or a deck.gl JSON spec dict."),
    ] = "html",
    title: Annotated[
        str | SkipJsonSchema[None],
        AdvancedField(
            default="",
            description="""\
            The map title. Note this is the title drawn on the map canvas itself, and will result
            in duplicate titles if set in the context of a dashboard in which the iframe/widget
            container also has a title set on it.
            """,
        ),
    ] = None,
    legend_style: Annotated[
        LegendStyle | SkipJsonSchema[None],
        AdvancedField(
            default=LegendStyle(),
            description="Additional arguments for configuring the legend.",
        ),
    ] = None,
    max_zoom: Annotated[
        int,
        AdvancedField(
            default=20,
            description="""\
            The maximum zoom level allowed by the map.
            This setting will be overridden if provided
            tile layers max zoom levels are lower than this value.
            """,
        ),
    ] = 20,
    view_state: Annotated[
        ViewState | SkipJsonSchema[None],
        AdvancedField(
            default=ViewState(),
            description="Manually set the view state of the map.",
        ),
    ] = None,
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str | DeckJsonSpec, Field()]:
    """
    Creates a map based on the provided layer definitions and configuration.

    Args:
    geo_layers (PydeckLayerDefinition | list[PydeckLayerDefinition] | None): Map layers to add to the map.
    tile_layers (list): A named tile layer, ie OpenStreetMap.
    static (bool): Set to true to disable map pan/zoom.
    title (str): The map title.
    legend_style (WidgetStyleBase): Additional arguments for configuring the Legend.
    max_zoom (int): The maximum zoom level of the map
    view_state (ViewState): Manually set the view state of the map, overrides any layer zoom settings.
    widget_id (str): The id of the dashboard widget that this tile layer belongs to.
        If set this MUST match the widget title as defined downstream in create_widget tasks

    Returns:
    str | DeckJsonSpec: A static HTML representation of the map, or a validated
        deck.gl JSON spec if ``output_type="json"``.
    """
    pdk.settings.custom_libraries = PYDECK_CUSTOM_LIBRARIES

    DEFAULT_WIDGETS = [
        pdk.Widget(
            "NorthArrowWidget",
            placement="top-left",
            style={"transform": "scale(0.8)"},
        ),
        pdk.Widget("ScaleWidget", placement="bottom-left"),
        pdk.Widget("SaveImageWidget", placement="top-right"),
    ]

    if tile_layers is None:
        tile_layers = []
    else:
        tile_layers = list(tile_layers)
    if legend_style is None:
        legend_style = LegendStyle()

    legend_values: list = []
    map_layers: list = []
    map_widgets: list = DEFAULT_WIDGETS.copy()

    for tile_layer in tile_layers:
        if isinstance(tile_layer, BitmapLayerDefinition):
            dump = _model_dump_for_pydeck(tile_layer)
            dump.pop("legend", None)
            layer = pdk.Layer("BitmapLayer", **dump)
            map_layers.append(layer)
            if tile_layer.legend is not None:
                legend_values.append(tile_layer.legend)
        else:
            layer = pdk.Layer(
                "TiledBitmapLayer",
                data=tile_layer.url,
                max_zoom=tile_layer.max_zoom,
                min_zoom=tile_layer.min_zoom,
                opacity=tile_layer.opacity,
                tile_size=256,
                widget_id=pdk.types.String(widget_id),
            )
            map_layers.append(layer)
            if tile_layer.max_zoom < max_zoom:
                max_zoom = tile_layer.max_zoom

    # Normalize geo_layers to a list
    if geo_layers is None:
        geo_layers = []
    elif isinstance(geo_layers, PydeckLayerDefinition):
        geo_layers = [geo_layers]

    tooltip_layer_columns: dict[str, list[str]] = {}
    any_pickable_geo_layer = False
    for layer_index, layer_def in enumerate(geo_layers):
        # Rendering: prefer data_url if set, fall back to geodataframe
        if layer_def.data_url is not None:
            data = pdk.types.String(layer_def.data_url)
        elif layer_def.geodataframe is not None:
            gdf = layer_def.geodataframe.to_crs("EPSG:4326")  # type: ignore[operator]
            # Pydeck's PolygonLayer does not support MultiPolygon geometries,
            # so we explode them into individual Polygons.
            is_multi = gdf.geometry.geom_type == "MultiPolygon"
            if is_multi.any():
                gdf = pd.concat(
                    [gdf[~is_multi], gdf[is_multi].explode(index_parts=False)],
                    ignore_index=True,
                )
            data = gdf

        layer_id = f"{layer_def.layer_type}-{layer_index}"
        style_dump = _model_dump_for_pydeck(layer_def.layer_style, layer_def.layer_type)
        layer = pdk.Layer(
            type=layer_def.layer_type,
            id=layer_id,
            data=data,
            **style_dump,
        )
        map_layers.append(layer)

        if layer_def.tooltip_columns is not None:
            tooltip_layer_columns[layer_id] = layer_def.tooltip_columns
        if getattr(layer_def.layer_style, "pickable", False):
            any_pickable_geo_layer = True

        # Legend: use geodataframe if present (regardless of rendering path)
        if legend_def := layer_def.legend:
            if isinstance(legend_def, LegendSegment):
                legend_values.append(legend_def)
            elif isinstance(legend_def, LegendFromDataframe):
                if layer_def.geodataframe is not None:
                    legend_values.append(legend_def.build_legend_from_dataframe(layer_def.geodataframe))
                else:
                    logger.warning(
                        "LegendFromDataframe legend skipped for layer '%s': "
                        "no geodataframe is available (layer uses data_url). "
                        "Use a LegendSegment to define a static legend for URL-backed layers.",
                        layer_def.layer_type,
                    )

    if legend_values:
        if legend_style.format_title:
            legend_values = [
                LegendSegment(title=legend_style.display_name(lv.title), values=lv.values) for lv in legend_values
            ]
        map_widgets.append(
            pdk.Widget(
                "LegendWidget",
                legend_values=legend_values,
                placement=legend_style.placement,
            )
        )

    if title:
        map_widgets.append(
            pdk.Widget(
                "TitleWidget",
                title=title,
            )
        )

    if any_pickable_geo_layer:
        map_widgets.append(
            pdk.Widget(
                "TooltipWidget",
                id="TooltipWidget",
                layer_columns=tooltip_layer_columns,
            )
        )

    if view_state is None:
        zoom_layers = [layer for layer in geo_layers if layer.zoom]
        if not zoom_layers:
            zoom_layers = geo_layers
        view_state = view_state_from_layers(layers=zoom_layers, max_zoom=max_zoom)

    m = pdk.Deck(
        layers=map_layers,
        widgets=map_widgets,
        initial_view_state=view_state,
        # The only non-default value here is repeat=True
        # which in our case allows tile layers to repeat/wrap at high zoom levels
        views=pdk.View(
            "MapView",
            controller=not static,
            repeat=True,
        ),
        # In order to avoid issues with z-fighting,
        # explicitly disable depth testing when layers have no extrusions
        parameters={"depthTest": any([getattr(layer, "extruded", False) for layer in map_layers])},
        map_style=pdk.map_styles.LIGHT_NO_LABELS,
    )

    if output_type == "json":
        return DeckJsonSpec.model_validate(json.loads(m.to_json()))
    return m.to_html(as_string=True)

draw_pie_chart 

draw_pie_chart(dataframe: DataFrame[JsonSerializableDataFrameModel], value_column: Annotated[str, Field(description='The name of the dataframe column to pull slice values from.')], label_column: Annotated[str | SkipJsonSchema[None], AdvancedField(default=None, description='The name of the dataframe column to label slices with, required if the data in value_column is numeric.')] = None, color_column: Annotated[str | SkipJsonSchema[None], AdvancedField(default=None, description='The name of the dataframe column to color slices with.')] = None, plot_style: Annotated[PlotStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional style kwargs passed to go.Pie().')] = None, layout_style: Annotated[LayoutStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs passed to plotly.go.Figure(layout).')] = None, widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Generates a pie chart from the provided params

Args: dataframe (pd.DataFrame): The input dataframe. value_column (str): The name of the dataframe column to pull slice values from. label_column (str): The name of the dataframe column to label slices with, required if the data in value_column is numeric. plot_style (PlotStyle): Additional style kwargs passed to go.Pie(). layout_style (LayoutStyle): Additional kwargs passed to plotly.go.Figure(layout). widget_id (str): The id of the dashboard widget that this tile layer belongs to. If set this MUST match the widget title as defined downstream in create_widget tasks

Returns: The generated chart html as a string

Source code in ecoscope/platform/tasks/results/_ecoplot.py 
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
@register()
def draw_pie_chart(
    dataframe: DataFrame[JsonSerializableDataFrameModel],
    value_column: Annotated[
        str,
        Field(description="The name of the dataframe column to pull slice values from."),
    ],
    label_column: Annotated[
        str | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description=(
                "The name of the dataframe column to label slices with,"
                " required if the data in value_column is numeric."
            ),
        ),
    ] = None,
    color_column: Annotated[
        str | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="The name of the dataframe column to color slices with.",
        ),
    ] = None,
    plot_style: Annotated[
        PlotStyle | SkipJsonSchema[None],
        AdvancedField(default=None, description="Additional style kwargs passed to go.Pie()."),
    ] = None,
    layout_style: Annotated[
        LayoutStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs passed to plotly.go.Figure(layout).",
        ),
    ] = None,
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Generates a pie chart from the provided params

    Args:
    dataframe (pd.DataFrame): The input dataframe.
    value_column (str): The name of the dataframe column to pull slice values from.
    label_column (str): The name of the dataframe column to label slices with,
        required if the data in value_column is numeric.
    plot_style (PlotStyle): Additional style kwargs passed to go.Pie().
    layout_style (LayoutStyle): Additional kwargs passed to plotly.go.Figure(layout).
    widget_id (str): The id of the dashboard widget that this tile layer belongs to.
        If set this MUST match the widget title as defined downstream in create_widget tasks

    Returns:
    The generated chart html as a string
    """
    from ecoscope.plotting import pie_chart

    plot = pie_chart(
        data=dataframe,
        value_column=value_column,
        label_column=label_column,
        color_column=color_column,
        style_kwargs=plot_style.model_dump(exclude_none=True) if plot_style else {},
        layout_kwargs=layout_style.model_dump(exclude_none=True) if layout_style else {},
    )

    return plot.to_html(**ExportArgs(div_id=widget_id).model_dump(exclude_none=True))

draw_table 

draw_table(dataframe: Annotated[AnyDataFrame, Field(description='The dataframe to render as a table.', exclude=True)], columns: Annotated[list[str] | SkipJsonSchema[None], AdvancedField(description='The list of dataframe columns to render in the table. Leave empty to render all columns', default=None)] = None, table_config: Annotated[TableConfig | SkipJsonSchema[None], AdvancedField(description='Configuration options for the table.', default=None)] = None, widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this table belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Creates an HTML table of the provided dataframe.

Source code in ecoscope/platform/tasks/results/_table.py 
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@register()
def draw_table(
    dataframe: Annotated[
        AnyDataFrame,
        Field(description="The dataframe to render as a table.", exclude=True),
    ],
    columns: Annotated[
        list[str] | SkipJsonSchema[None],
        AdvancedField(
            description="The list of dataframe columns to render in the table. Leave empty to render all columns",
            default=None,
        ),
    ] = None,
    table_config: Annotated[
        TableConfig | SkipJsonSchema[None],
        AdvancedField(
            description="Configuration options for the table.",
            default=None,
        ),
    ] = None,
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this table belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Creates an HTML table of the provided dataframe.
    """
    import geopandas as gpd  # type: ignore[import-untyped]

    if table_config is None:
        table_config = TableConfig()

    if isinstance(dataframe, gpd.GeoDataFrame):
        dataframe = dataframe.drop(columns="geometry")
        dataframe = pd.DataFrame(dataframe)  # type: ignore[assignment]

    dataframe = dataframe[columns] if columns else dataframe
    dataframe = _convert_json_columns_to_string(dataframe)

    table_row_data = dataframe.to_json(orient="records", date_format="iso")
    table_column_defs = json.dumps([{"field": col, "headerTooltip": col} for col in dataframe.columns])

    return HTML_TEMPLATE.format(
        table_row_data=table_row_data,
        table_column_defs=table_column_defs,
        table_config=table_config.model_dump_json(),
        widget_id=widget_id,
    )

draw_time_series_bar_chart 

draw_time_series_bar_chart(dataframe: DataFrame[JsonSerializableDataFrameModel], x_axis: Annotated[str, Field(description='The dataframe column to plot in the x/time axis.')], y_axis: Annotated[str, Field(description='The dataframe column to plot in the y axis.')], category: Annotated[str, Field(description='The dataframe column to stack in the y axis.')], agg_function: Annotated[AggOperations, Field(description='The aggregate function to apply to the group.')], time_interval: Annotated[Literal['year', 'month', 'week', 'day', 'hour'], Field()], color_column: Annotated[str | SkipJsonSchema[None], AdvancedField(default=None, description='The name of the dataframe column to color bars with.')] = None, plot_style: Annotated[PlotStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional style kwargs passed to go.Bar().')] = None, layout_style: Annotated[BarLayoutStyle | SkipJsonSchema[None], AdvancedField(default=None, description='Additional kwargs passed to plotly.go.Figure(layout).')] = None, widget_id: Annotated[str | SkipJsonSchema[None], Field(description='            The id of the dashboard widget that this tile layer belongs to.\n            If set this MUST match the widget title as defined downstream in create_widget tasks\n            ', exclude=True)] = None) -> Annotated[str, Field()]

Generates a stacked time series bar chart from the provided params

Args: dataframe (pd.DataFrame): The input dataframe. x_axis (str): The dataframe column to plot in the x axis. y_axis (str): The dataframe column to plot in the y axis. category (str): The dataframe column to stack in the y axis. agg_function (str): The aggregate function to apply to the group. time_interval (str): Sets the time interval of the x axis. color_column (str): The name of the dataframe column to color bars with. plot_style (PlotStyle): Style arguments passed to plotly.graph_objects.Bar and applied to all groups. layout_style (LayoutStyle): Additional kwargs passed to plotly.go.Figure(layout). widget_id (str): The id of the dashboard widget that this tile layer belongs to. If set this MUST match the widget title as defined downstream in create_widget tasks

Returns: The generated chart html as a string

Source code in ecoscope/platform/tasks/results/_ecoplot.py 
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
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
@register()
def draw_time_series_bar_chart(
    dataframe: DataFrame[JsonSerializableDataFrameModel],
    x_axis: Annotated[str, Field(description="The dataframe column to plot in the x/time axis.")],
    y_axis: Annotated[str, Field(description="The dataframe column to plot in the y axis.")],
    category: Annotated[str, Field(description="The dataframe column to stack in the y axis.")],
    agg_function: Annotated[
        AggOperations,
        Field(description="The aggregate function to apply to the group."),
    ],
    time_interval: Annotated[
        Literal["year", "month", "week", "day", "hour"],
        Field(),
    ],
    color_column: Annotated[
        str | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="The name of the dataframe column to color bars with.",
        ),
    ] = None,
    plot_style: Annotated[
        PlotStyle | SkipJsonSchema[None],
        AdvancedField(default=None, description="Additional style kwargs passed to go.Bar()."),
    ] = None,
    layout_style: Annotated[
        BarLayoutStyle | SkipJsonSchema[None],
        AdvancedField(
            default=None,
            description="Additional kwargs passed to plotly.go.Figure(layout).",
        ),
    ] = None,
    widget_id: Annotated[
        str | SkipJsonSchema[None],
        Field(
            description="""\
            The id of the dashboard widget that this tile layer belongs to.
            If set this MUST match the widget title as defined downstream in create_widget tasks
            """,
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field()]:
    """
    Generates a stacked time series bar chart from the provided params

    Args:
    dataframe (pd.DataFrame): The input dataframe.
    x_axis (str): The dataframe column to plot in the x axis.
    y_axis (str): The dataframe column to plot in the y axis.
    category (str): The dataframe column to stack in the y axis.
    agg_function (str): The aggregate function to apply to the group.
    time_interval (str): Sets the time interval of the x axis.
    color_column (str): The name of the dataframe column to color bars with.
    plot_style (PlotStyle): Style arguments passed to plotly.graph_objects.Bar and applied to all groups.
    layout_style (LayoutStyle): Additional kwargs passed to plotly.go.Figure(layout).
    widget_id (str): The id of the dashboard widget that this tile layer belongs to.
        If set this MUST match the widget title as defined downstream in create_widget tasks

    Returns:
    The generated chart html as a string
    """
    import datetime

    from ecoscope.plotting import EcoPlotData, stacked_bar_chart

    layout_kws = layout_style.model_dump(exclude_none=True) if layout_style else {}
    plot_style = plot_style if plot_style else PlotStyle()

    match time_interval:
        case "year":
            dataframe["truncated_time"] = dataframe[x_axis].apply(lambda x: datetime.datetime(x.year, 1, 1))
            layout_kws["xaxis_dtick"] = "M12"
        case "month":
            dataframe["truncated_time"] = dataframe[x_axis].apply(lambda x: datetime.datetime(x.year, x.month, 1))
            layout_kws["xaxis_dtick"] = "M1"
            plot_style.width = MONTH_IN_MILLISECONDS
            plot_style.xperiod = "M1"
            plot_style.xperiodalignment = "start"
        case "week":
            dataframe["truncated_time"] = dataframe[x_axis].apply(
                lambda x: datetime.datetime(x.year, x.month, x.day) - datetime.timedelta(x.day_of_week)
            )
            layout_kws["xaxis_dtick"] = WEEK_IN_MILLISECONDS
        case "day":
            dataframe["truncated_time"] = dataframe[x_axis].apply(lambda x: datetime.datetime(x.year, x.month, x.day))
            layout_kws["xaxis_dtick"] = DAY_IN_MILLISECONDS
        case "hour":
            dataframe["truncated_time"] = dataframe[x_axis].apply(
                lambda x: datetime.datetime(x.year, x.month, x.day, x.hour)
            )
            layout_kws["xaxis_dtick"] = HOUR_IN_MILLISECONDS
        case _:
            raise NotImplementedError(f"Unsupported time_interval: {time_interval}")

    grouped = dataframe.groupby(["truncated_time", category])

    data = EcoPlotData(
        grouped=grouped,
        x_col="truncated_time",
        y_col=y_axis,
        color_col=color_column,
        **(plot_style.model_dump(exclude_none=True) if plot_style else {}),
    )

    plot = stacked_bar_chart(
        data=data,
        agg_function=agg_function,
        stack_column=category,
        layout_kwargs=layout_kws,
    )

    return plot.to_html(**ExportArgs(div_id=widget_id).model_dump(exclude_none=True))

gather_dashboard 

gather_dashboard(details: Annotated[WorkflowDetails, Field(description='Workflow details')], widgets: Annotated[NestedWidgetList | FlatWidgetList | GroupedOrSingleWidget, Field(description='The widgets to display.', exclude=True)], groupers: Annotated[AllGrouper | list[ValueGrouper | TemporalGrouper | SpatialGrouper] | SkipJsonSchema[None], Field(description='            Groupers that are used to group the widgets.\n            If all widgets are ungrouped, this field defaults to `None`.\n            ', exclude=True)] = None, time_range: Annotated[TimeRange | SkipJsonSchema[None], Field(description='Time range filter')] = None, warning: Annotated[str | SkipJsonSchema[None], Field(exclude=True)] = None) -> Annotated[Dashboard, Field()]
Source code in ecoscope/platform/tasks/results/_dashboard.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
@register()
def gather_dashboard(
    details: Annotated[WorkflowDetails, Field(description="Workflow details")],
    widgets: Annotated[
        NestedWidgetList | FlatWidgetList | GroupedOrSingleWidget,
        Field(description="The widgets to display.", exclude=True),
    ],
    groupers: Annotated[
        AllGrouper | list[ValueGrouper | TemporalGrouper | SpatialGrouper] | SkipJsonSchema[None],
        Field(
            description="""\
            Groupers that are used to group the widgets.
            If all widgets are ungrouped, this field defaults to `None`.
            """,
            exclude=True,
        ),
    ] = None,
    time_range: Annotated[TimeRange | SkipJsonSchema[None], Field(description="Time range filter")] = None,
    warning: Annotated[str | SkipJsonSchema[None], Field(exclude=True)] = None,
) -> Annotated[Dashboard, Field()]:
    # if the input is any kind of list, try to flatten it because it might be nested
    # if not a list, make it a single-element list to allow uniform handling below
    as_flat_list = _flatten(widgets) if isinstance(widgets, list) else [widgets]
    # then regardless of element type(s), parse to uniform flat list of GroupedWidgets
    grouped_widgets = [
        GroupedWidget.from_single_view(w) if isinstance(w, WidgetSingleView) else w for w in as_flat_list
    ]
    if groupers:
        all_view_keys = set(
            [vk for view_keys in [list(gw.views) for gw in grouped_widgets] for vk in view_keys if vk is not None],
        )
        for gw in grouped_widgets:
            if list(gw.views) == [None]:
                # this widget is meant to have only one view,
                # so we don't need to add null views for it
                continue
            for missing_view_key in all_view_keys.difference(gw.views):
                gw.add_null_view(missing_view_key)
        for gw in grouped_widgets:
            if list(gw.views) != [None]:
                # now make sure all grouped widgets have the same keys.
                assert set(list(gw.views)) == all_view_keys, "All grouped widgets must have the same keys"
        grouper_choices = composite_filters_to_grouper_choices_dict(groupers, list(all_view_keys))
    formatted_time_range = ""
    time_zone_label = ""
    if time_range:
        formatted_time_range = (
            f"From {time_range.since.strftime(time_range.time_format)}"
            f" to {time_range.until.strftime(time_range.time_format)}"
        )
        time_zone_label = time_range.timezone.label

    return Dashboard(
        widgets=grouped_widgets,
        grouper_choices=(grouper_choices if groupers else None),
        keys=(sorted(list(all_view_keys)) if groupers else None),
        metadata=Metadata(
            title=details.name,
            description=details.description,
            time_range=formatted_time_range,
            time_zone=time_zone_label,
            warning=warning,
        ),
    )

gather_output_files 

gather_output_files(files: Annotated[list[str | list[tuple[CompositeFilter, str]]], Field(description='The files to gather.', exclude=True)]) -> OutputFiles

Gather the output files from the tasks.

Parameters:

Name Type Description Default
files Annotated[list[str | list[tuple[CompositeFilter, str]]], Field(description='The files to gather.', exclude=True)]

A list of files to gather output files from.

 required

Returns:

Type Description
OutputFiles

A list of output files.
Source code in ecoscope/platform/tasks/results/_output_files.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@register()
def gather_output_files(
    files: Annotated[
        list[str | list[tuple[CompositeFilter, str]]],
        Field(description="The files to gather.", exclude=True),
    ],
) -> OutputFiles:
    """Gather the output files from the tasks.

    Args:
        files: A list of files to gather output files from.

    Returns:
        A list of output files.
    """
    return OutputFiles(files=files)

merge_tile_layers 

merge_tile_layers(base_layers: Annotated[list[TileLayer] | SkipJsonSchema[None], Field(description='Static base tile layers to prepend.')] = None, overlay: Annotated[BitmapLayerDefinition | SkipJsonSchema[None], Field(description='Per-group overlay tile layer to append.')] = None) -> Annotated[list[TileLayer | BitmapLayerDefinition], Field()]

Merges static base tile layers with a per-group overlay into a single list.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
@register()
def merge_tile_layers(
    base_layers: Annotated[
        list[TileLayer] | SkipJsonSchema[None],
        Field(description="Static base tile layers to prepend."),
    ] = None,
    overlay: Annotated[
        BitmapLayerDefinition | SkipJsonSchema[None],
        Field(description="Per-group overlay tile layer to append."),
    ] = None,
) -> Annotated[list[TileLayer | BitmapLayerDefinition], Field()]:
    """Merges static base tile layers with a per-group overlay into a single list."""
    layers: list[TileLayer | BitmapLayerDefinition] = []
    if base_layers:
        layers.extend(base_layers)
    if overlay:
        layers.append(overlay)
    return layers

merge_widget_views 

merge_widget_views(widgets: Annotated[list[WidgetSingleView], Field(description='The widgets to merge', exclude=True)]) -> Annotated[list[GroupedWidget], Field(description='The merged widgets')]

Merge widgets with the same title and widget_type.

Parameters:

Name Type Description Default
widgets Annotated[list[WidgetSingleView], Field(description='The widgets to merge', exclude=True)]

The widgets to merge.

 required

Returns:

Type Description
Annotated[list[GroupedWidget], Field(description='The merged widgets')]

The merged grouped widgets.
Source code in ecoscope/platform/tasks/results/_widget_tasks.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
@register()
def merge_widget_views(
    widgets: Annotated[
        list[WidgetSingleView],
        Field(description="The widgets to merge", exclude=True),
    ],
) -> Annotated[list[GroupedWidget], Field(description="The merged widgets")]:
    """Merge widgets with the same `title` and `widget_type`.

    Args:
        widgets: The widgets to merge.

    Returns:
        The merged grouped widgets.
    """
    grouped_widgets = [GroupedWidget.from_single_view(w) for w in widgets]
    merged: dict[GroupedWidgetMergeKey, GroupedWidget] = {}
    for gw in grouped_widgets:
        if gw.merge_key not in merged:
            merged[gw.merge_key] = gw
        else:
            merged[gw.merge_key] |= gw
    return list(merged.values())

persist_geoarrow_for_pydeck 

persist_geoarrow_for_pydeck(gdf: Annotated[AnyGeoDataFrame, Field(description='GeoDataframe to persist as GeoArrow-encoded parquet')], root_path: Annotated[str, Field(description='Root path to persist parquet to')], filename: Annotated[str | None, Field(description='Optional filename within `root_path`. Auto-generated from a df content hash if absent. The `.parquet` extension is appended automatically.', exclude=True)] = None) -> Annotated[str, Field(description='Path to persisted parquet')]

Persist a gdf as GeoArrow-encoded parquet, ready for use in geoarrow layers.

Intended for use with the maps created by this module, use tasks.io.persist_df(filetype='geoparquet') instead when writing for standard WKB-expecting consumers (e.g. QGIS, PostGIS)

Source code in ecoscope/platform/tasks/results/_map_utils.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
@register()
def persist_geoarrow_for_pydeck(
    gdf: Annotated[AnyGeoDataFrame, Field(description="GeoDataframe to persist as GeoArrow-encoded parquet")],
    root_path: Annotated[str, Field(description="Root path to persist parquet to")],
    filename: Annotated[
        str | None,
        Field(
            description=(
                "Optional filename within `root_path`. Auto-generated from a "
                "df content hash if absent. The `.parquet` extension is "
                "appended automatically."
            ),
            exclude=True,
        ),
    ] = None,
) -> Annotated[str, Field(description="Path to persisted parquet")]:
    """Persist a gdf as GeoArrow-encoded parquet, ready for
    use in geoarrow layers.

    Intended for use with the maps created by this module, use
    `tasks.io.persist_df(filetype='geoparquet')` instead when writing for
    standard WKB-expecting consumers (e.g. QGIS, PostGIS)
    """

    if not filename:
        try:
            hash_input = bytes(pd.util.hash_pandas_object(gdf).values)
        except (TypeError, ValueError):
            hash_input = f"{gdf.shape}{gdf.head(5).to_dict()}".encode()
        filename = hashlib.sha256(hash_input).hexdigest()[:7]

    gdf = gpd.GeoDataFrame(gdf).copy()
    _iso_format_timestamp_columns(gdf)
    _downcast_float_columns(gdf)
    _pack_color_columns(gdf)
    _stringify_mixed_json(gdf)
    buffer = io.BytesIO()
    gdf.to_parquet(buffer, index=False, geometry_encoding="geoarrow")
    return _persist_bytes(buffer.getvalue(), root_path, f"{filename}.parquet")

rewrite_file_urls_for_screenshots 

rewrite_file_urls_for_screenshots(html: Annotated[str, Field(description='HTML string output from draw_map.')], file_urls: Annotated[list[str], Field(description='The file url strings to replace in `html`.')]) -> Annotated[str, Field()]

Rewrites file_urls in map HTML to http://127.0.0.1:<port>/ so that Playwright can fetch local files without CORS restrictions when serve_local_files=True is set on ScreenshotConfig.

Only the filename (stem + extension) is preserved — the full local path is dropped. For example file:///some/long/path/data.geojson becomes http://127.0.0.1:8099/data.geojson.

The port defaults to 8099 and can be overridden via the ECOSCOPE_SCREENSHOT_FILE_SERVER_PORT environment variable.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
@register()
def rewrite_file_urls_for_screenshots(
    html: Annotated[str, Field(description="HTML string output from draw_map.")],
    file_urls: Annotated[list[str], Field(description="The file url strings to replace in `html`.")],
) -> Annotated[str, Field()]:
    """
    Rewrites file_urls in map HTML to ``http://127.0.0.1:<port>/``
    so that Playwright can fetch local files without CORS restrictions when
    ``serve_local_files=True`` is set on ``ScreenshotConfig``.

    Only the filename (stem + extension) is preserved — the full local path is
    dropped. For example ``file:///some/long/path/data.geojson`` becomes
    ``http://127.0.0.1:8099/data.geojson``.

    The port defaults to ``8099`` and can be overridden via the
    ``ECOSCOPE_SCREENSHOT_FILE_SERVER_PORT`` environment variable.
    """
    import os

    port = int(os.environ.get("ECOSCOPE_SCREENSHOT_FILE_SERVER_PORT", 8099))
    base_url = f"http://127.0.0.1:{port}"

    for file_url in file_urls:
        filename = os.path.basename(file_url)
        html = html.replace(file_url, f"{base_url}/{filename}")

    return html

set_base_maps 

set_base_maps(base_maps: Annotated[list[TileLayer] | SkipJsonSchema[None], Field(json_schema_extra=_preset_or_custom_json_schema_extra, title=' ', description='Select tile layers to use as base layers in map outputs. The first layer in the list will be the bottommost layer displayed.')] = None) -> Annotated[list[TileLayer], Field()]
Source code in ecoscope/platform/tasks/results/_map_utils.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
@register()
def set_base_maps(
    base_maps: Annotated[
        list[TileLayer] | SkipJsonSchema[None],
        Field(
            json_schema_extra=_preset_or_custom_json_schema_extra,
            title=" ",
            description=(
                "Select tile layers to use as base layers in map outputs."
                " The first layer in the list will be the bottommost layer displayed."
            ),
        ),
    ] = None,
) -> Annotated[list[TileLayer], Field()]:
    if base_maps is None:
        base_maps = [
            TileLayer(layer_name="TERRAIN"),
            TileLayer(layer_name="SATELLITE", opacity=0.5),
        ]
    return base_maps

set_layer_opacity 

set_layer_opacity(opacity: OpacityAnnotation = 1) -> float
Source code in ecoscope/platform/tasks/results/_map_utils.py 
30
31
32
33
34
@register()
def set_layer_opacity(
    opacity: OpacityAnnotation = 1,
) -> float:
    return opacity

shift_radius_values 

shift_radius_values(gdf: Annotated[AnyGeoDataFrame, Field(description='Source geodataframe.', exclude=True)], radius_column: Annotated[str, Field(description='Column name whose values will be lifted/imputed for use as a scatterplot radius accessor.')]) -> Annotated[AnyGeoDataFrame, Field()]

Shifts the given radius column so that values are >= 1 and NaNs are replaced with 1. Intended for use when displaying the given gdf as a scatterplot layer, where the numeric values in radius_column determine the size of the scatterplot points.

Source code in ecoscope/platform/tasks/results/_pydeck.py 
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
@register()
def shift_radius_values(
    gdf: Annotated[
        AnyGeoDataFrame,
        Field(description="Source geodataframe.", exclude=True),
    ],
    radius_column: Annotated[
        str,
        Field(
            description=(
                "Column name whose values will be lifted/imputed for use as a " "scatterplot radius accessor."
            ),
        ),
    ],
) -> Annotated[AnyGeoDataFrame, Field()]:
    """
    Shifts the given radius column so that values are >= 1 and NaNs are replaced with 1.
    Intended for use when displaying the given gdf as a scatterplot layer, where
    the numeric values in radius_column determine the size of the scatterplot points.
    """
    series = gdf[radius_column]
    # Lift all values up such that the min == 1
    if series.min() < 0:
        series = series + (0 - series.min()) + 1
    # set nans to 1, and lift everything else by 1 to distinguish NaN's and minimums
    if series.hasnans:
        series = series + 1
        series = series.fillna(1)
    gdf[radius_column] = series
    return gdf

view_state_from_geodataframes 

view_state_from_geodataframes(geodataframes: list[AnyGeoDataFrame], max_zoom: float = 20) -> Annotated[ViewState, Field()]
Source code in ecoscope/platform/tasks/results/_pydeck.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
@register()
def view_state_from_geodataframes(
    geodataframes: list[AnyGeoDataFrame],
    max_zoom: float = 20,
) -> Annotated[ViewState, Field()]:
    if not geodataframes:
        return ViewState()

    bounds = pd.concat(geodataframes).total_bounds
    bbox = [
        [bounds[0], bounds[1]],  # Northwest corner
        [bounds[2], bounds[3]],  # Southeast corner
    ]
    computed_zoom = pdk.data_utils.viewport_helpers.bbox_to_zoom_level(bbox)
    center_lon = (bounds[0] + bounds[2]) / 2
    center_lat = (bounds[1] + bounds[3]) / 2

    return ViewState(longitude=center_lon, latitude=center_lat, zoom=min(max_zoom, computed_zoom))

view_state_from_layers 

view_state_from_layers(layers: list[PydeckLayerDefinition], max_zoom: float = 20) -> Annotated[ViewState, Field()]
Source code in ecoscope/platform/tasks/results/_pydeck.py 
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
@register()
def view_state_from_layers(
    layers: list[PydeckLayerDefinition],
    max_zoom: float = 20,
) -> Annotated[ViewState, Field()]:
    # TODO: handle view_state for 3-d layers with elevation
    gdfs = [layer.geodataframe for layer in layers if layer.geodataframe is not None]
    if not gdfs:
        logger.warning(
            "view_state_from_layers: no geodataframe data available (all layers use data_url). "
            "Falling back to default view state centred on (0, 0). "
            "Pass an explicit view_state to draw_map to override this."
        )
    return view_state_from_geodataframes(
        geodataframes=gdfs,
        max_zoom=max_zoom,
    )