Skip to content

Filter Tasks

ecoscope.platform.tasks.filter

Attributes

UTC_TIMEZONEINFO module-attribute 

UTC_TIMEZONEINFO = TimezoneInfo(label='UTC', tzCode='UTC', name='UTC', utc_offset='+00:00')

Classes

TimeRange

Bases: BaseModel

Attributes
since instance-attribute 
since: datetime
time_format class-attribute instance-attribute 
time_format: str = DEFAULT_TIME_FORMAT
timezone instance-attribute 
timezone: TimezoneInfo
until instance-attribute 
until: datetime
Methods:
ensure_timezone_awareness 
ensure_timezone_awareness() -> TimeRange
Source code in ecoscope/platform/tasks/filter/_filter.py 
36
37
38
39
40
41
42
43
44
45
46
@model_validator(mode="after")
def ensure_timezone_awareness(self) -> "TimeRange":
    both_naive = not self.since.tzinfo and not self.until.tzinfo
    both_aware = self.since.tzinfo and self.until.tzinfo
    if not both_naive and not both_aware:
        raise ValueError("Since and until values must both be timezone naive, or both aware")
    if both_naive:
        tz = self.timezone.utc_offset_as_datetime_timezone
        self.since = self.since.replace(tzinfo=tz)
        self.until = self.until.replace(tzinfo=tz)
    return self

TimezoneInfo

Bases: BaseModel

Attributes
label instance-attribute 
label: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
name instance-attribute 
name: str
tzCode instance-attribute 
tzCode: str
utc_offset instance-attribute 
utc_offset: Annotated[str, Field(alias=utc)]
utc_offset_as_datetime_timezone property 
utc_offset_as_datetime_timezone

Functions:

get_timezone_from_time_range 

get_timezone_from_time_range(time_range: TimeRange) -> Annotated[TimezoneInfo, Field()]

Utility function to return the TimezoneInfo object nested in the provided TimeRange in workflow specs

Source code in ecoscope/platform/tasks/filter/_filter.py 
86
87
88
89
90
91
92
93
@register()
def get_timezone_from_time_range(
    time_range: TimeRange,
) -> Annotated[TimezoneInfo, Field()]:
    """
    Utility function to return the TimezoneInfo object nested in the provided TimeRange in workflow specs
    """
    return time_range.timezone

set_time_range 

set_time_range(since: Annotated[datetime, Field(description='The start time')], until: Annotated[datetime, Field(description='The end time')], timezone: Annotated[TimezoneInfo | SkipJsonSchema[None], Field(default=None)] = None, time_format: Annotated[str, AdvancedField(default=DEFAULT_TIME_FORMAT, description='The time format')] = DEFAULT_TIME_FORMAT) -> Annotated[TimeRange, Field(description='Time range filter')]
Source code in ecoscope/platform/tasks/filter/_filter.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@register(description="Choose the period of time to analyze.")
def set_time_range(
    since: Annotated[datetime, Field(description="The start time")],
    until: Annotated[datetime, Field(description="The end time")],
    timezone: Annotated[
        TimezoneInfo | SkipJsonSchema[None],
        Field(default=None),
    ] = None,
    time_format: Annotated[
        str,
        AdvancedField(
            default=DEFAULT_TIME_FORMAT,
            description="The time format",
        ),
    ] = DEFAULT_TIME_FORMAT,
) -> Annotated[TimeRange, Field(description="Time range filter")]:
    if timezone is None:
        # Assume UTC if no timezone is provided
        timezone = UTC_TIMEZONEINFO
    return TimeRange(since=since, until=until, timezone=timezone, time_format=time_format)