Skip to content

Config Tasks

ecoscope.platform.tasks.config

Classes

WorkflowDetails dataclass 

WorkflowDetails(name: str, description: str, image_url: str = '')
Attributes
description instance-attribute 
description: str
image_url class-attribute instance-attribute 
image_url: str = ''
name instance-attribute 
name: str

Functions:

call_etd_from_combined_params 

call_etd_from_combined_params(trajectory_gdf: TrajectoryAnnotation, combined_params: EtdArgsWithOpacity) -> DataFrame[TimeDensityReturnGDFSchema]
Source code in ecoscope/platform/tasks/config/_meta_tasks.py 
114
115
116
117
118
119
120
121
122
123
@register()
def call_etd_from_combined_params(
    trajectory_gdf: TrajectoryAnnotation,
    combined_params: EtdArgsWithOpacity,
) -> DataFrame[TimeDensityReturnGDFSchema]:
    return (
        task(calculate_elliptical_time_density)
        .validate()
        .call(trajectory_gdf=trajectory_gdf, **combined_params.get_etd_params())
    )

call_ltd_from_combined_params 

call_ltd_from_combined_params(trajectory_gdf: TrajectoryAnnotation, meshgrid: MeshGridAnnotation, combined_params: LtdArgsWithOpacity) -> AnyGeoDataFrame
Source code in ecoscope/platform/tasks/config/_meta_tasks.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@register()
def call_ltd_from_combined_params(
    trajectory_gdf: TrajectoryAnnotation,
    meshgrid: MeshGridAnnotation,
    combined_params: LtdArgsWithOpacity,
) -> AnyGeoDataFrame:
    return (
        task(calculate_linear_time_density)
        .validate()
        .call(
            trajectory_gdf=trajectory_gdf,
            meshgrid=meshgrid,
            **combined_params.get_ltd_params(),
        )
    )

call_meshgrid_from_combined_params 

call_meshgrid_from_combined_params(aoi: AoiAnnotation, combined_params: LtdArgsWithOpacity) -> AnyGeoDataFrame
Source code in ecoscope/platform/tasks/config/_meta_tasks.py 
126
127
128
129
130
131
@register()
def call_meshgrid_from_combined_params(
    aoi: AoiAnnotation,
    combined_params: LtdArgsWithOpacity,
) -> AnyGeoDataFrame:
    return task(create_meshgrid).validate().call(aoi=aoi, **combined_params.get_meshgrid_params())

concat_string_vars 

concat_string_vars(values: Annotated[list[str | SkipSentinel], Field(description='The values to concatenate')]) -> str
Source code in ecoscope/platform/tasks/config/_set_vars.py 
55
56
57
58
59
60
@register()
def concat_string_vars(
    values: Annotated[list[str | SkipSentinel], Field(description="The values to concatenate")],
) -> str:
    values_to_concat: list[str] = [v for v in values if not isinstance(v, SkipSentinel)]
    return "".join(values_to_concat)

default_if_string_is_empty 

default_if_string_is_empty(value: Annotated[str, Field(description='The value to passthrough')], default: Annotated[str | None | SkipSentinel, Field(description='Default if `value` is None')]) -> str | None | SkipSentinel
Source code in ecoscope/platform/tasks/config/_set_vars.py 
47
48
49
50
51
52
@register()
def default_if_string_is_empty(
    value: Annotated[str, Field(description="The value to passthrough")],
    default: Annotated[str | None | SkipSentinel, Field(description="Default if `value` is None")],
) -> str | None | SkipSentinel:
    return default if value == "" else value

default_if_string_is_none_or_skip 

default_if_string_is_none_or_skip(value: Annotated[str | None | SkipSentinel, Field(description='The value to passthrough')], default: Annotated[str, Field(description='Default if `value` is None or Skip')]) -> str
Source code in ecoscope/platform/tasks/config/_set_vars.py 
39
40
41
42
43
44
@register()
def default_if_string_is_none_or_skip(
    value: Annotated[str | None | SkipSentinel, Field(description="The value to passthrough")],
    default: Annotated[str, Field(description="Default if `value` is None or Skip")],
) -> str:
    return default if value is None or isinstance(value, SkipSentinel) else value

get_column_names_from_dataframe 

get_column_names_from_dataframe(df: Annotated[AnyDataFrame, Field(...)], exclude_column_names: Annotated[list[str] | None, Field(...)]) -> list[str]
Source code in ecoscope/platform/tasks/config/_set_vars.py 
70
71
72
73
74
75
76
77
78
79
80
@register()
def get_column_names_from_dataframe(
    df: Annotated[AnyDataFrame, Field(...)],
    exclude_column_names: Annotated[list[str] | None, Field(...)],
) -> list[str]:
    columns = df.columns.to_list()

    if exclude_column_names:
        columns = [col for col in columns if col not in exclude_column_names]

    return columns

get_opacity_from_combined_params 

get_opacity_from_combined_params(combined_params: EtdArgsWithOpacity | LtdArgsWithOpacity) -> float
Source code in ecoscope/platform/tasks/config/_meta_tasks.py 
151
152
153
154
155
@register()
def get_opacity_from_combined_params(
    combined_params: EtdArgsWithOpacity | LtdArgsWithOpacity,
) -> float:
    return combined_params.opacity

prefix_string_var 

prefix_string_var(var: Annotated[str, Field(title='')], prefix: Annotated[str, Field(title='')]) -> str
Source code in ecoscope/platform/tasks/config/_set_vars.py 
31
32
33
34
35
36
@register()
def prefix_string_var(
    var: Annotated[str, Field(title="")],
    prefix: Annotated[str, Field(title="")],
) -> str:
    return f"{prefix}{var}"

set_bool_var 

set_bool_var(var: Annotated[bool, Field(title='')]) -> bool
Source code in ecoscope/platform/tasks/config/_set_vars.py 
17
18
19
20
21
@register()
def set_bool_var(
    var: Annotated[bool, Field(title="")],
) -> bool:
    return var

set_etd_args_with_opacity 

set_etd_args_with_opacity(opacity: OpacityAnnotation, auto_scale_or_custom_cell_size: AutoScaleOrCustomAnnotation = None, crs: CrsAnnotation = 'EPSG:3857', nodata_value: NoDataAnnotation = 'nan', band_count: BandCountAnnotation = 1, max_speed_factor: MaxSpeedFactorAnnotation = 1.05, expansion_factor: ExpansionFactorAnnotation = 1.3, percentiles: EtdPercentileAnnotation = None) -> EtdArgsWithOpacity
Source code in ecoscope/platform/tasks/config/_meta_tasks.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@register()
def set_etd_args_with_opacity(
    opacity: OpacityAnnotation,
    auto_scale_or_custom_cell_size: AutoScaleOrCustomAnnotation = None,
    crs: CrsAnnotation = "EPSG:3857",
    nodata_value: NoDataAnnotation = "nan",
    band_count: BandCountAnnotation = 1,
    max_speed_factor: MaxSpeedFactorAnnotation = 1.05,
    expansion_factor: ExpansionFactorAnnotation = 1.3,
    percentiles: EtdPercentileAnnotation = None,
) -> EtdArgsWithOpacity:
    return EtdArgsWithOpacity(
        opacity=opacity,
        auto_scale_or_custom_cell_size=auto_scale_or_custom_cell_size,
        crs=crs,
        nodata_value=nodata_value,
        band_count=band_count,
        max_speed_factor=max_speed_factor,
        expansion_factor=expansion_factor,
        percentiles=percentiles,
    )

set_list_of_string_vars 

set_list_of_string_vars(vars: Annotated[list[str], Field(title='')]) -> list[str]
Source code in ecoscope/platform/tasks/config/_set_vars.py 
24
25
26
27
28
@register()
def set_list_of_string_vars(
    vars: Annotated[list[str], Field(title="")],
) -> list[str]:
    return vars

set_ltd_args_with_opacity 

set_ltd_args_with_opacity(opacity: OpacityAnnotation, auto_scale_or_custom_cell_size: AutoScaleOrCustomAnnotation = None, crs: CrsAnnotation = 'EPSG:3857', intersecting_only: IntersectingOnlyAnnotation = False, percentiles: LtdPercentileAnnotation = None) -> LtdArgsWithOpacity
Source code in ecoscope/platform/tasks/config/_meta_tasks.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@register()
def set_ltd_args_with_opacity(
    opacity: OpacityAnnotation,
    auto_scale_or_custom_cell_size: AutoScaleOrCustomAnnotation = None,
    crs: CrsAnnotation = "EPSG:3857",
    intersecting_only: IntersectingOnlyAnnotation = False,
    percentiles: LtdPercentileAnnotation = None,
) -> LtdArgsWithOpacity:
    return LtdArgsWithOpacity(
        opacity=opacity,
        auto_scale_or_custom_cell_size=auto_scale_or_custom_cell_size,
        crs=crs,
        intersecting_only=intersecting_only,
        percentiles=percentiles,
    )

set_string_var 

set_string_var(var: Annotated[str, Field(title='')]) -> str
Source code in ecoscope/platform/tasks/config/_set_vars.py 
10
11
12
13
14
@register()
def set_string_var(
    var: Annotated[str, Field(title="")],
) -> str:
    return var

set_workflow_details 

set_workflow_details(name: Annotated[str, Field(title='Workflow Name')], description: Annotated[str, Field(title='Workflow Description', default='')] = '', image_url: Annotated[str, Field(description='An image url', default='', exclude=True)] = '') -> WorkflowDetails
Source code in ecoscope/platform/tasks/config/_workflow_details.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
@register(description="Add information that will help to differentiate this workflow from another.")
def set_workflow_details(
    name: Annotated[str, Field(title="Workflow Name")],
    description: Annotated[str, Field(title="Workflow Description", default="")] = "",
    image_url: Annotated[
        str, Field(description="An image url", default="", exclude=True)
    ] = "",  # This is excluded due to https://github.com/wildlife-dynamics/compose/issues/308
) -> WorkflowDetails:
    return WorkflowDetails(
        name=name,
        description=description,
        image_url=image_url,
    )

title_case_var 

title_case_var(var: Annotated[str, Field(...)]) -> str
Source code in ecoscope/platform/tasks/config/_set_vars.py 
63
64
65
66
67
@register()
def title_case_var(
    var: Annotated[str, Field(...)],
) -> str:
    return var.replace("_", " ").title()