CountEmbedder

srai.embedders.CountEmbedder

CountEmbedder(
    expected_output_features: Optional[
        Union[list[str], OsmTagsFilter, GroupedOsmTagsFilter]
    ] = None,
    count_subcategories: bool = True,
)

Bases: Embedder

Simple Embedder that counts occurences of feature values.

PARAMETER	DESCRIPTION
count_subcategories	Whether to count all subcategories individually or count features only on the highest level based on features column name. Defaults to True. TYPE: bool DEFAULT: True

Source code in srai/embedders/count_embedder.py

def __init__(
    self,
    expected_output_features: Optional[
        Union[list[str], OsmTagsFilter, GroupedOsmTagsFilter]
    ] = None,
    count_subcategories: bool = True,
) -> None:
    """
    Init CountEmbedder.

    Args:
        expected_output_features
            (Union[List[str], OsmTagsFilter, GroupedOsmTagsFilter], optional):
            The features that are expected to be found in the resulting embedding.
            If not None, the missing features are added and filled with 0.
            The unexpected features are removed. The resulting columns are sorted accordingly.
            Defaults to None.
        count_subcategories (bool, optional): Whether to count all subcategories individually
            or count features only on the highest level based on features column name.
            Defaults to True.
    """
    self.count_subcategories = count_subcategories
    self._parse_expected_output_features(expected_output_features)

transform

transform(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> pd.DataFrame

Embed a given GeoDataFrame.

Creates region embeddings by counting the frequencies of each feature value. Expects features_gdf to be in wide format with each column being a separate type of feature (e.g. amenity, leisure) and rows to hold values of these features for each object. The resulting DataFrame will have columns made by combining the feature name (column) and value (row) e.g. amenity_fuel or type_0. The rows will hold numbers of this type of feature in each region.

PARAMETER	DESCRIPTION
regions_gdf	Region indexes and geometries. TYPE: GeoDataFrame
features_gdf	Feature indexes, geometries and feature values. TYPE: GeoDataFrame
joint_gdf	Joiner result with region-feature multi-index. TYPE: GeoDataFrame

RETURNS	DESCRIPTION
DataFrame	pd.DataFrame: Embedding for each region in regions_gdf.

RAISES	DESCRIPTION
ValueError	If features_gdf is empty and self.expected_output_features is not set.
ValueError	If any of the gdfs index names is None.
ValueError	If joint_gdf.index is not of type pd.MultiIndex or doesn't have 2 levels.
ValueError	If index levels in gdfs don't overlap correctly.
ValueError	If features_gdf contains boolean columns and count_subcategories is True.

Source code in srai/embedders/count_embedder.py

def transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> pd.DataFrame:
    """
    Embed a given GeoDataFrame.

    Creates region embeddings by counting the frequencies of each feature value.
    Expects features_gdf to be in wide format with each column
    being a separate type of feature (e.g. amenity, leisure)
    and rows to hold values of these features for each object.
    The resulting DataFrame will have columns made by combining
    the feature name (column) and value (row) e.g. amenity_fuel or type_0.
    The rows will hold numbers of this type of feature in each region.

    Args:
        regions_gdf (gpd.GeoDataFrame): Region indexes and geometries.
        features_gdf (gpd.GeoDataFrame): Feature indexes, geometries and feature values.
        joint_gdf (gpd.GeoDataFrame): Joiner result with region-feature multi-index.

    Returns:
        pd.DataFrame: Embedding for each region in regions_gdf.

    Raises:
        ValueError: If features_gdf is empty and self.expected_output_features is not set.
        ValueError: If any of the gdfs index names is None.
        ValueError: If joint_gdf.index is not of type pd.MultiIndex or doesn't have 2 levels.
        ValueError: If index levels in gdfs don't overlap correctly.
        ValueError: If features_gdf contains boolean columns and count_subcategories is True.
    """
    self._validate_indexes(regions_gdf, features_gdf, joint_gdf)
    if features_gdf.empty:
        if self.expected_output_features is not None:
            return pd.DataFrame(
                0, index=regions_gdf.index, columns=self.expected_output_features
            )
        else:
            raise ValueError(
                "Cannot embed with empty features_gdf and no expected_output_features."
            )

    regions_df = pl.from_arrow(
        pa.Table.from_pandas(
            regions_gdf[[]],
        ),
        schema={REGIONS_INDEX: pl.String()},
    ).lazy()
    features_df = pl.from_arrow(
        pa.Table.from_pandas(
            features_gdf,
            columns=[c for c in features_gdf.columns if c != GEOMETRY_COLUMN],
        )
    ).lazy()
    joint_df = pl.from_arrow(
        pa.Table.from_pandas(
            joint_gdf[[]],
        ),
        schema={REGIONS_INDEX: pl.String(), FEATURES_INDEX: pl.String()},
    ).lazy()

    features_schema = features_df.collect_schema()
    feature_columns = [col for col in features_schema.names() if col != FEATURES_INDEX]
    dtypes = features_schema.dtypes()
    are_all_columns_bool = all(
        dtypes[idx] == pl.Boolean
        for idx, col in enumerate(features_schema.names())
        if col != FEATURES_INDEX
    )

    if self.count_subcategories:
        if are_all_columns_bool:
            raise ValueError("Cannot count subcategories with boolean columns.")

        feature_encodings = (
            features_df.collect(streaming=True).to_dummies(columns=feature_columns).lazy()
        )
        feature_columns = [
            col
            for col in feature_encodings.collect_schema().names()
            if col != FEATURES_INDEX and not col.endswith("_null")
        ]
        feature_encodings = feature_encodings.select([FEATURES_INDEX, *feature_columns])
    elif are_all_columns_bool:
        feature_encodings = features_df.with_columns(
            [
                pl.col(FEATURES_INDEX),
                *(pl.col(col).cast(pl.Int32) for col in feature_columns),
            ]
        )
    else:
        feature_encodings = features_df.with_columns(
            [
                pl.col(FEATURES_INDEX),
                *(pl.col(col).is_not_null().cast(pl.Int32) for col in feature_columns),
            ]
        )

    joint_with_encodings = joint_df.join(feature_encodings, on=FEATURES_INDEX, how="left")
    region_embeddings = joint_with_encodings.drop(FEATURES_INDEX).group_by(REGIONS_INDEX).sum()
    region_embeddings, feature_columns = self._maybe_filter_to_expected_features(
        region_embeddings, feature_columns
    )

    region_embeddings_df = (
        (
            regions_df.join(region_embeddings, on=REGIONS_INDEX, how="left")
            .fill_null(0)
            .with_columns(
                [
                    pl.col(REGIONS_INDEX),
                    *(pl.col(col).cast(pl.Int32) for col in feature_columns),
                ]
            )
        )
        .collect(streaming=True)
        .to_pandas()
        .set_index(REGIONS_INDEX)
    )

    return region_embeddings_df