connections

ecoscope.platform.connections

Attributes

ClientProtocolType module-attribute

ClientProtocolType = TypeVar('ClientProtocolType')

DataConnectionType module-attribute

DataConnectionType = TypeVar('DataConnectionType', bound='DataConnection')

EarthEngineClient module-attribute

EarthEngineClient = Annotated[EarthEngineClientProtocol, BeforeValidator(client_from_named_connection), WithJsonSchema({'type': 'string', 'description': 'A named Google EarthEngine connection.'})]

EarthRangerClient module-attribute

EarthRangerClient = Annotated[EarthRangerClientProtocol, BeforeValidator(client_from_named_connection), WithJsonSchema({'type': 'string', 'description': 'A named EarthRanger connection.'})]

SmartClient module-attribute

SmartClient = Annotated[SmartClientProtocol, BeforeValidator(client_from_named_connection), WithJsonSchema({'type': 'string', 'description': 'A named SMART connection.'})]

Classes

DataConnection

Bases: ABC, _DataConnection, Generic[ClientProtocolType]

Methods:

client_from_named_connection classmethod

client_from_named_connection(name: str) -> ClientProtocolType

Source code in ecoscope/platform/connections.py

@classmethod
def client_from_named_connection(cls, name: str) -> ClientProtocolType:
    return cls.from_named_connection(name).get_client()

get_client abstractmethod

get_client() -> ClientProtocolType

Source code in ecoscope/platform/connections.py

@abstractmethod
def get_client(self) -> ClientProtocolType: ...

EarthEngineClientProtocol

Bases: Protocol

EarthEngineConnection

Bases: DataConnection[EarthEngineClientProtocol]

Attributes

ee_project class-attribute instance-attribute

ee_project: Annotated[str, Field(description='Your EarthEngine project ID')] = ''

private_key class-attribute instance-attribute

private_key: Annotated[SecretStr, Field(description="Your service account's private key")] = SecretStr('')

private_key_file class-attribute instance-attribute

private_key_file: Annotated[str, Field(description="Your service account's private key")] = ''

service_account class-attribute instance-attribute

service_account: Annotated[str, Field(description='Your Google Cloud Service Account')] = ''

Methods:

get_client

get_client()

Source code in ecoscope/platform/connections.py

def get_client(self):
    from ecoscope.io import EarthEngineIO

    return EarthEngineIO(
        service_account=self.service_account,
        private_key=self.private_key.get_secret_value(),
        private_key_file=self.private_key_file,
        ee_project=self.ee_project,
    )

EarthRangerClientProtocol

Bases: Protocol

Attributes

server instance-attribute

server: str

token instance-attribute

token: str | None

Methods:

get_choices_from_v2_event_type

get_choices_from_v2_event_type(event_type, choice_field) -> dict[str, str]

Source code in ecoscope/platform/connections.py

def get_choices_from_v2_event_type(self, event_type, choice_field) -> dict[str, str]: ...

get_event_type_display_names_from_events

get_event_type_display_names_from_events(events_gdf, append_category_names) -> AnyGeoDataFrame

Source code in ecoscope/platform/connections.py

def get_event_type_display_names_from_events(
    self,
    events_gdf,
    append_category_names,
) -> AnyGeoDataFrame: ...

get_event_types

get_event_types() -> AnyDataFrame

Source code in ecoscope/platform/connections.py

def get_event_types(self) -> AnyDataFrame: ...

get_events

get_events(since, until, event_type, drop_null_geometry, include_details: bool, include_updates: bool, include_related_events: bool, force_point_geometry: bool) -> AnyGeoDataFrame

Source code in ecoscope/platform/connections.py

def get_events(
    self,
    since,
    until,
    event_type,
    drop_null_geometry,
    include_details: bool,
    include_updates: bool,
    include_related_events: bool,
    force_point_geometry: bool,
) -> AnyGeoDataFrame: ...

get_fields_from_event_type_schema

get_fields_from_event_type_schema(event_type) -> dict[str, str]

Source code in ecoscope/platform/connections.py

def get_fields_from_event_type_schema(self, event_type) -> dict[str, str]: ...

get_patrol_events

get_patrol_events(since, until, patrol_type_value, event_type, status, drop_null_geometry, sub_page_size, patrols_overlap_daterange) -> AnyGeoDataFrame

Source code in ecoscope/platform/connections.py

def get_patrol_events(
    self,
    since,
    until,
    patrol_type_value,
    event_type,
    status,
    drop_null_geometry,
    sub_page_size,
    patrols_overlap_daterange,
) -> AnyGeoDataFrame: ...

get_patrol_observations

get_patrol_observations(patrols_df, include_patrol_details, sub_page_size) -> AnyGeoDataFrame

Source code in ecoscope/platform/connections.py

def get_patrol_observations(
    self,
    patrols_df,
    include_patrol_details,
    sub_page_size,
) -> AnyGeoDataFrame: ...

get_patrol_observations_with_patrol_filter

get_patrol_observations_with_patrol_filter(since, until, patrol_type_value, status, include_patrol_details, sub_page_size, patrols_overlap_daterange) -> AnyGeoDataFrame

Source code in ecoscope/platform/connections.py

def get_patrol_observations_with_patrol_filter(
    self,
    since,
    until,
    patrol_type_value,
    status,
    include_patrol_details,
    sub_page_size,
    patrols_overlap_daterange,
) -> AnyGeoDataFrame: ...

get_patrols

get_patrols(since, until, patrol_type_value, status, sub_page_size, patrols_overlap_daterange) -> AnyDataFrame

Source code in ecoscope/platform/connections.py

def get_patrols(
    self,
    since,
    until,
    patrol_type_value,
    status,
    sub_page_size,
    patrols_overlap_daterange,
) -> AnyDataFrame: ...

get_spatial_features_group

get_spatial_features_group(spatial_features_group_name, spatial_features_group_id, with_group_data) -> dict[str, str | int | AnyGeoDataFrame]

Source code in ecoscope/platform/connections.py

def get_spatial_features_group(
    self,
    spatial_features_group_name,
    spatial_features_group_id,
    with_group_data,
) -> dict[str, str | int | AnyGeoDataFrame]: ...

get_subjectgroup_observations

get_subjectgroup_observations(subject_group_name: str, include_subject_details: bool, include_inactive: bool, include_details: bool, include_subjectsource_details: bool, since, until, filter) -> AnyGeoDataFrame

Source code in ecoscope/platform/connections.py

def get_subjectgroup_observations(
    self,
    subject_group_name: str,
    include_subject_details: bool,
    include_inactive: bool,
    include_details: bool,
    include_subjectsource_details: bool,
    since,
    until,
    filter,
) -> AnyGeoDataFrame: ...

EarthRangerConnection

Bases: DataConnection[EarthRangerClientProtocol]

Attributes

password class-attribute instance-attribute

password: Annotated[SecretStr, Field(description='EarthRanger password')] = SecretStr('')

server instance-attribute

server: Annotated[str, Field(description='EarthRanger API URL')]

sub_page_size class-attribute instance-attribute

sub_page_size: Annotated[int, Field(description='Sub page size for API requests')] = 4000

tcp_limit class-attribute instance-attribute

tcp_limit: Annotated[int, Field(description='TCP limit for API requests')] = 5

token class-attribute instance-attribute

token: Annotated[SecretStr, Field(description='EarthRanger access token')] = SecretStr('')

username class-attribute instance-attribute

username: Annotated[str, Field(description='EarthRanger username')] = ''

Methods:

get_client

get_client() -> EarthRangerClientProtocol

Source code in ecoscope/platform/connections.py

def get_client(self) -> EarthRangerClientProtocol:
    from ecoscope.io import EarthRangerIO

    auth_kws = (
        {"token": self.token.get_secret_value()}
        if self.token
        else {
            "username": self.username,
            "password": self.password.get_secret_value(),
        }
    )
    return EarthRangerIO(
        server=self.server,
        tcp_limit=self.tcp_limit,
        sub_page_size=self.sub_page_size,
        **auth_kws,
    )

token_or_password

token_or_password(v: SecretStr, info: ValidationInfo)

Source code in ecoscope/platform/connections.py

@field_validator("token")
def token_or_password(cls, v: SecretStr, info: ValidationInfo):
    if v and (info.data["username"] or info.data["password"]):
        raise ValueError("Only a token, or an EarthRanger username and password can be provided, not both")
    if not v and not (info.data["username"] and info.data["password"]):
        raise ValueError("If token is empty, EarthRanger username and password must be provided")
    return v

SmartClientProtocol

Bases: Protocol

Methods:

get_events

get_events(ca_uuid: str, language_uuid: str, start: str, end: str) -> AnyGeoDataFrame

Source code in ecoscope/platform/connections.py

def get_events(self, ca_uuid: str, language_uuid: str, start: str, end: str) -> AnyGeoDataFrame: ...

get_patrol_observations

get_patrol_observations(ca_uuid: str, language_uuid: str, start: str, end: str, patrol_mandate: str | None = ..., patrol_transport: str | None = ..., station_uuid: str | None = ...) -> Any

Source code in ecoscope/platform/connections.py

def get_patrol_observations(
    self,
    ca_uuid: str,
    language_uuid: str,
    start: str,
    end: str,
    patrol_mandate: str | None = ...,
    patrol_transport: str | None = ...,
    station_uuid: str | None = ...,
) -> Any: ...

SmartConnection

Bases: DataConnection[SmartClientProtocol]

Attributes

password class-attribute instance-attribute

password: Annotated[SecretStr, Field(description='Smart password')] = SecretStr('')

server instance-attribute

server: Annotated[str, Field(description='Smart API URL')]

token class-attribute instance-attribute

token: Annotated[SecretStr, Field(description='Smart access token')] = SecretStr('')

username class-attribute instance-attribute

username: Annotated[str, Field(description='Smart username')] = ''

Methods:

get_client

get_client() -> SmartClientProtocol

Source code in ecoscope/platform/connections.py

def get_client(self) -> SmartClientProtocol:
    from ecoscope.io import SmartIO

    auth_kws = (
        {"token": self.token.get_secret_value()}
        if self.token
        else {
            "username": self.username,
            "password": self.password.get_secret_value(),
        }
    )
    return SmartIO(
        urlBase=self.server,
        **auth_kws,
    )

token_or_password

token_or_password(v: SecretStr, info: ValidationInfo)

Source code in ecoscope/platform/connections.py

@field_validator("token")
def token_or_password(cls, v: SecretStr, info: ValidationInfo):
    if v and (info.data["username"] or info.data["password"]):
        raise ValueError("Only a token, or an EarthRanger username and password can be provided, not both")
    if not v and not (info.data["username"] and info.data["password"]):
        raise ValueError("If token is empty, SMART username and password must be provided")
    return v

Functions:

connection_from_client

connection_from_client(obj) -> DataConnection

Source code in ecoscope/platform/connections.py

def connection_from_client(obj) -> DataConnection:
    assert is_client(obj)
    bv = [arg for arg in get_args(obj) if isinstance(arg, BeforeValidator)][0]
    conn_type = bv.func.__self__  # type: ignore[union-attr]
    assert issubclass(conn_type, DataConnection)
    return conn_type

is_client

is_client(obj)

Source code in ecoscope/platform/connections.py

def is_client(obj):
    if hasattr(obj, "__origin__") and hasattr(obj, "__args__"):
        if any(isinstance(arg, BeforeValidator) for arg in get_args(obj)):
            bv = [arg for arg in get_args(obj) if isinstance(arg, BeforeValidator)][0]
            if ismethod(bv.func) and bv.func.__name__ == "client_from_named_connection":
                return True
    return False