annotations

ecoscope.platform.annotations

Attributes

AdvancedField module-attribute

AdvancedField = create_custom_field('AdvancedField', always_set={'json_schema_extra': {'ecoscope:advanced': True}}, require_default=True)

AnyDataFrame module-attribute

AnyDataFrame = DataFrame[JsonSerializableDataFrameModel]

AnyDataFrameOrEmpty module-attribute

AnyDataFrameOrEmpty = Union[AnyDataFrame, EmptyDataFrame]

AnyGeoDataFrame module-attribute

AnyGeoDataFrame = DataFrame[GeoDataFrameBaseSchema]

AnyGeoDataFrameOrEmpty module-attribute

AnyGeoDataFrameOrEmpty = Union[AnyGeoDataFrame, EmptyDataFrame]

DataFrame module-attribute

DataFrame = Annotated[DataFrame[DataFrameSchema], WithJsonSchema({'type': 'ecoscope.platform.annotations.DataFrame'})]

DataFrameSchema module-attribute

DataFrameSchema = TypeVar('DataFrameSchema', bound=JsonSerializableDataFrameModel)

EmptyDataFrame module-attribute

EmptyDataFrame = DataFrame[EmptyDataFrameModel]

FP module-attribute

FP = ParamSpec('FP')

Classes

EmptyDataFrameModel

Bases: JsonSerializableDataFrameModel

Methods:

is_empty

is_empty(df: DataFrame)

Source code in ecoscope/platform/annotations.py

@pa.dataframe_check
def is_empty(cls, df: pd.DataFrame):  # type: ignore[misc]
    return df.empty

GeoDataFrameBaseSchema

Bases: JsonSerializableDataFrameModel

Attributes

geometry class-attribute instance-attribute

geometry: Series[Any] = Field(nullable=True)

JsonSerializableDataFrameModel

Bases: DataFrameModel

StrictGeoDataFrameBaseSchema

Bases: JsonSerializableDataFrameModel

Attributes

geometry class-attribute instance-attribute

geometry: Series[Any] = Field(nullable=False)

Functions:

create_custom_field

create_custom_field(field_name: str, always_set: dict, require_default: bool, *, __f: Callable[FP, FieldInfo] = Field) -> Callable[FP, FieldInfo]

A factory function which creates a custom Field with always_set kwargs always set. Note that the always_set kwargs will not be omitted from the signature of the returned custom Field as understood by static type checkers, but they will raise ValueError at runtime if overridden. If require_default is True, the user must provide a default when calling the custom Field.

Parameters:

Name	Type	Description	Default
always_set	dict	The keyword arguments to always set on Field.	required
require_default	bool	If True, the user must provide a default when calling the custom Field.	required
__f	Callable[FP, FR]	Defaults to Field. This is included in the signature for editor support only; the default should not be overridden.	Field

Source code in ecoscope/platform/annotations.py

def create_custom_field(
    field_name: str,
    always_set: dict,
    require_default: bool,
    *,
    __f: Callable[FP, FieldInfo] = Field,  # type: ignore[assignment]
) -> Callable[FP, FieldInfo]:
    """A factory function which creates a custom Field with `always_set` kwargs always set.
    Note that the `always_set` kwargs will not be omitted from the signature of the returned
    custom Field as understood by static type checkers, but they will raise `ValueError` at
    runtime if overridden. If `require_default` is `True`, the user must provide a default when
    calling the custom Field.

    Args:
        always_set (dict): The keyword arguments to always set on Field.
        require_default (bool): If True, the user must provide a default when calling
            the custom Field.
        __f (Callable[FP, FR], optional): Defaults to Field. This is included in the signature
            for editor support only; the default should not be overridden.
    """
    if not __f == Field:
        raise ValueError(
            "This factory only supports `pydantic.Field` as the value for `__f`; `__f` "
            "is included in the signature for editor support only; the default should "
            "not be overridden."
        )

    def wrapper(*args, **kwargs):
        # Take a copy of always_set since we may want to mutate it
        fixed_kwargs = always_set.copy()
        if require_default and (not args and "default" not in kwargs):
            raise ValueError(f"A default is required for fields constructed with '{field_name}'.")
        # In the specific case of json_schema_extra, we want to merge anything in
        # always_set into the user provided kwargs, and raise on any conflicts
        if always_json_schema_extra := always_set.get("json_schema_extra", False):
            user_json_schema_extra = kwargs.pop("json_schema_extra", {})
            if any(key in user_json_schema_extra for key in always_json_schema_extra):
                raise ValueError(
                    f"Fields constructed with '{field_name}' cannot override "
                    f"json_schema_extra: {always_json_schema_extra.keys()}"
                )
            fixed_kwargs["json_schema_extra"] = user_json_schema_extra | always_json_schema_extra
        if any(key in kwargs for key in always_set):
            raise ValueError(f"Fields constructed with '{field_name}' cannot override: {always_set.keys()}")
        kwargs.update(fixed_kwargs)
        return __f(*args, **kwargs)

    return wrapper

is_subscripted_pandera_dataframe

is_subscripted_pandera_dataframe(obj)

Source code in ecoscope/platform/annotations.py

def is_subscripted_pandera_dataframe(obj):
    if hasattr(obj, "__origin__") and hasattr(obj, "__args__"):
        if get_origin(obj) == pa_typing.DataFrame:
            return True
    return False