Index

embedders ¶

This module contains embedders, used to convert spatial data to their vector representations.

Embedders are designed to unify different types of spatial data embedding methods, such as hex2vec or gtfs2vec into a single interface. This allows to easily switch between different embedding methods without changing the rest of the code.

ContextualCountEmbedder ¶

ContextualCountEmbedder(
    neighbourhood: Neighbourhood[IndexType],
    neighbourhood_distance: int,
    concatenate_vectors: bool = False,
    expected_output_features: Optional[
        Union[list[str], OsmTagsFilter, GroupedOsmTagsFilter]
    ] = None,
    count_subcategories: bool = False,
    aggregation_function: Literal[
        "average", "median", "sum", "min", "max"
    ] = "average",
    num_of_multiprocessing_workers: int = -1,
    multiprocessing_activation_threshold: Optional[int] = None,
)

Bases: CountEmbedder

ContextualCountEmbedder.

PARAMETER	DESCRIPTION
`neighbourhood`	Neighbourhood object used to get neighbours for the contextualization. TYPE: `Neighbourhood[T]`
`neighbourhood_distance`	How many neighbours levels should be included in the embedding. TYPE: `int`
`concatenate_vectors`	Whether to sum all neighbours into a single vector with the same width as `CountEmbedder`, or to concatenate them to the wide format and keep all neighbour levels separate. Defaults to False. TYPE: `bool` DEFAULT: `False`
`count_subcategories`	Whether to count all subcategories individually or count features only on the highest level based on features column name. Defaults to False. TYPE: `bool` DEFAULT: `False`
`aggregation_function`	Function used to aggregate data from the neighbours. Defaults to average. TYPE: `Literal['average', 'median', 'sum', 'min', 'max']` DEFAULT: `'average'`
`num_of_multiprocessing_workers`	Number of workers used for multiprocessing. Defaults to -1 which results in a total number of available cpu threads. `0` and `1` values disable multiprocessing. Similar to `n_jobs` parameter from `scikit-learn` library. TYPE: `int` DEFAULT: `-1`
`multiprocessing_activation_threshold`	Number of seeds required to start processing on multiple processes. Activating multiprocessing for a small amount of points might not be feasible. Defaults to 100. TYPE: `int` DEFAULT: `None`

RAISES	DESCRIPTION
`ValueError`	If `neighbourhood_distance` is negative.

Source code in srai/embedders/contextual_count_embedder.py

def __init__(
    self,
    neighbourhood: Neighbourhood[IndexType],
    neighbourhood_distance: int,
    concatenate_vectors: bool = False,
    expected_output_features: Optional[
        Union[list[str], OsmTagsFilter, GroupedOsmTagsFilter]
    ] = None,
    count_subcategories: bool = False,
    aggregation_function: Literal["average", "median", "sum", "min", "max"] = "average",
    num_of_multiprocessing_workers: int = -1,
    multiprocessing_activation_threshold: Optional[int] = None,
) -> None:
    """
    Init ContextualCountEmbedder.

    Args:
        neighbourhood (Neighbourhood[T]): Neighbourhood object used to get neighbours for
            the contextualization.
        neighbourhood_distance (int): How many neighbours levels should be included in
            the embedding.
        concatenate_vectors (bool, optional): Whether to sum all neighbours into a single vector
            with the same width as `CountEmbedder`, or to concatenate them to the wide format
            and keep all neighbour levels separate. Defaults to False.
        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 False.
        aggregation_function (Literal["average", "median", "sum", "min", "max"], optional):
            Function used to aggregate data from the neighbours. Defaults to average.
        num_of_multiprocessing_workers (int, optional): Number of workers used for
            multiprocessing. Defaults to -1 which results in a total number of available
            cpu threads. `0` and `1` values disable multiprocessing.
            Similar to `n_jobs` parameter from `scikit-learn` library.
        multiprocessing_activation_threshold (int, optional): Number of seeds required to start
            processing on multiple processes. Activating multiprocessing for a small
            amount of points might not be feasible. Defaults to 100.

    Raises:
        ValueError: If `neighbourhood_distance` is negative.
    """
    super().__init__(expected_output_features, count_subcategories)

    self.neighbourhood = neighbourhood
    self.neighbourhood_distance = neighbourhood_distance
    self.concatenate_vectors = concatenate_vectors
    self.aggregation_function = aggregation_function

    if self.neighbourhood_distance < 0:
        raise ValueError("Neighbourhood distance must be positive.")

    self.num_of_multiprocessing_workers = _parse_num_of_multiprocessing_workers(
        num_of_multiprocessing_workers
    )
    self.multiprocessing_activation_threshold = _parse_multiprocessing_activation_threshold(
        multiprocessing_activation_threshold
    )

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 and applying a contextualization based on neighbours of regions. For each region, features will be altered based on the neighbours either by adding aggregated values dimished based on distance, or by adding new separate columns with neighbour distance postfix. 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 rows will hold numbers of this type of feature in each region. Numbers can be fractional because neighbourhoods are aggregated to represent a single value from all neighbours on a given level.

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.

Source code in srai/embedders/contextual_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 and applying
    a contextualization based on neighbours of regions. For each region, features will be
    altered based on the neighbours either by adding aggregated values dimished based on
    distance, or by adding new separate columns with neighbour distance postfix.
    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 rows will hold numbers of this type of feature in each region. Numbers can be
    fractional because neighbourhoods are aggregated to represent a single value from
    all neighbours on a given level.

    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.
    """
    counts_df = super().transform(regions_gdf, features_gdf, joint_gdf)

    result_df: pd.DataFrame
    if self.concatenate_vectors:
        result_df = self._get_concatenated_embeddings(counts_df)
    else:
        result_df = self._get_squashed_embeddings(counts_df)

    return result_df

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

Embedder ¶

Bases: ABC

Abstract class for embedders.

transform ¶

abstractmethod

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

Embed regions using features.

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 and geometry index for each region in regions_gdf.

RAISES	DESCRIPTION
`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.

Source code in srai/embedders/_base.py

@abc.abstractmethod
def transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> pd.DataFrame:  # pragma: no cover
    """
    Embed regions using features.

    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 and geometry index for each region in regions_gdf.

    Raises:
        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.
    """
    raise NotImplementedError

GTFS2VecEmbedder ¶

GTFS2VecEmbedder(
    hidden_size: int = 48,
    embedding_size: int = 64,
    skip_autoencoder: bool = False,
)

Bases: Embedder

GTFS2Vec Embedder.

PARAMETER	DESCRIPTION
`hidden_size`	Hidden size in encoder and decoder. Defaults to 48. TYPE: `int` DEFAULT: `48`
`embedding_size`	Embedding size. Defaults to 64. TYPE: `int` DEFAULT: `64`
`skip_autoencoder`	Skip using autoencoder as part of embedding. TYPE: `bool` DEFAULT: `False`

Source code in srai/embedders/gtfs2vec/embedder.py

def __init__(
    self, hidden_size: int = 48, embedding_size: int = 64, skip_autoencoder: bool = False
) -> None:
    """
    Init GTFS2VecEmbedder.

    Args:
        hidden_size (int, optional): Hidden size in encoder and decoder. Defaults to 48.
        embedding_size (int, optional): Embedding size. Defaults to 64.
        skip_autoencoder (bool, optional): Skip using autoencoder as part of embedding.
        Defaults to False.
    """
    import_optional_dependencies(
        dependency_group="torch", modules=["torch", "pytorch_lightning"]
    )
    self._model: Optional[GTFS2VecModel] = None
    self._hidden_size = hidden_size
    self._embedding_size = embedding_size
    self._skip_autoencoder = skip_autoencoder
    self._is_fitted = False

fit ¶

fit(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> None

Fit model to a given data.

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`

RAISES	DESCRIPTION
`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.

Source code in srai/embedders/gtfs2vec/embedder.py

def fit(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> None:
    """
    Fit model to a given data.

    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.

    Raises:
        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.
    """
    self._validate_indexes(regions_gdf, features_gdf, joint_gdf)
    features = self._prepare_features(regions_gdf, features_gdf, joint_gdf)

    if not self._skip_autoencoder:
        self._model = self._train_model_unsupervised(features)

fit_transform ¶

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

Fit model and transform a given data.

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 and geometry index for each region in regions_gdf.

RAISES	DESCRIPTION
`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.

Source code in srai/embedders/gtfs2vec/embedder.py

def fit_transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> pd.DataFrame:
    """
    Fit model and transform a given data.

    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 and geometry index for each region in regions_gdf.

    Raises:
        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.
    """
    self._validate_indexes(regions_gdf, features_gdf, joint_gdf)
    features = self._prepare_features(regions_gdf, features_gdf, joint_gdf)

    if self._skip_autoencoder:
        return features
    else:
        self._model = self._train_model_unsupervised(features)
        return self._embed(features)

load ¶

classmethod

load(path: Union[Path, str]) -> GTFS2VecEmbedder

Load the model from a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path`

RETURNS	DESCRIPTION
`Hex2VecEmbedder`	The loaded embedder. TYPE: `GTFS2VecEmbedder`

Source code in srai/embedders/gtfs2vec/embedder.py

@classmethod
def load(cls, path: Union[Path, str]) -> "GTFS2VecEmbedder":
    """
    Load the model from a directory.

    Args:
        path (Path): Path to the directory.

    Returns:
        Hex2VecEmbedder: The loaded embedder.
    """
    return cls._load(path, GTFS2VecModel)

save ¶

save(path: Union[Path, str]) -> None

Save the model to a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path`

Source code in srai/embedders/gtfs2vec/embedder.py

def save(self, path: Union[Path, str]) -> None:
    """
    Save the model to a directory.

    Args:
        path (Path): Path to the directory.
    """
    embedder_config = {
        "hidden_size": self._hidden_size,
        "embedding_size": self._embedding_size,
        "skip_autoencoder": self._skip_autoencoder,
    }
    self._save(path, embedder_config)

transform ¶

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

Embed a given data.

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 and geometry index for each region in regions_gdf.

RAISES	DESCRIPTION
`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 number of features is incosistent with the model.
`ModelNotFitException`	If model is not fit.

Source code in srai/embedders/gtfs2vec/embedder.py

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

    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 and geometry index for each region in regions_gdf.

    Raises:
        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 number of features is incosistent with the model.
        ModelNotFitException: If model is not fit.
    """
    self._validate_indexes(regions_gdf, features_gdf, joint_gdf)
    features = self._prepare_features(regions_gdf, features_gdf, joint_gdf)

    if self._skip_autoencoder:
        return features
    return self._embed(features)

GeoVexEmbedder ¶

GeoVexEmbedder(
    target_features: Union[list[str], OsmTagsFilter, GroupedOsmTagsFilter],
    count_subcategories: bool = True,
    batch_size: Optional[int] = 32,
    neighbourhood_radius: int = 4,
    convolutional_layers: int = 2,
    embedding_size: int = 32,
    convolutional_layer_size: int = 256,
)

Bases: CountEmbedder

GeoVex Embedder.

PARAMETER	DESCRIPTION
`target_features`	The features that are to be used in the embedding. Should be in "flat" format, i.e. "_", or use OsmTagsFilter object. TYPE: `Union[List[str], OsmTagsFilter, GroupedOsmTagsFilter]`
`count_subcategories`	Whether to count all subcategories individually or count features only on the highest level based on features column name. Defaults to False. TYPE: `bool` DEFAULT: `True`
`batch_size`	Batch size. Defaults to 32. TYPE: `int` DEFAULT: `32`
`convolutional_layers`	Number of convolutional layers. Defaults to 2. TYPE: `int` DEFAULT: `2`
`neighbourhood_radius`	Radius of the neighbourhood. Defaults to 4. TYPE: `int` DEFAULT: `4`
`embedding_size`	Size of the embedding. Defaults to 32. TYPE: `int` DEFAULT: `32`
`convolutional_layer_size`	Size of the first convolutional layer. TYPE: `int` DEFAULT: `256`

Source code in srai/embedders/geovex/embedder.py

def __init__(
    self,
    target_features: Union[list[str], OsmTagsFilter, GroupedOsmTagsFilter],
    count_subcategories: bool = True,
    batch_size: Optional[int] = 32,
    neighbourhood_radius: int = 4,
    convolutional_layers: int = 2,
    embedding_size: int = 32,
    convolutional_layer_size: int = 256,
) -> None:
    """
    Initialize GeoVex Embedder.

    Args:
        target_features (Union[List[str], OsmTagsFilter, GroupedOsmTagsFilter]): The features
            that are to be used in the embedding. Should be in "flat" format,
            i.e. "<super-tag>_<sub-tag>", or use OsmTagsFilter object.
        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 False.
        batch_size (int, optional): Batch size. Defaults to 32.
        convolutional_layers (int, optional): Number of convolutional layers. Defaults to 2.
        neighbourhood_radius (int, optional): Radius of the neighbourhood. Defaults to 4.
        embedding_size (int, optional): Size of the embedding. Defaults to 32.
        convolutional_layer_size (int, optional): Size of the first convolutional layer.
    """
    import_optional_dependencies(
        dependency_group="torch", modules=["torch", "pytorch_lightning"]
    )

    super().__init__(
        expected_output_features=target_features,
        count_subcategories=count_subcategories,
    )

    self._assert_feature_length(self.expected_output_features, convolutional_layer_size)

    self._model: Optional[GeoVexModel] = None
    self._is_fitted = False
    self._r = neighbourhood_radius
    self._embedding_size = embedding_size
    self._convolutional_layers = convolutional_layers
    self._convolutional_layer_size = convolutional_layer_size

    self._batch_size = batch_size

    # save invalid h3s for later
    self._invalid_cells: list[str] = []
    self._dataset: DataLoader = None

invalid_cells `property` ¶

invalid_cells: list[str]

List of invalid h3s.

fit ¶

fit(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: H3Neighbourhood,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> None

Fit the model to the data.

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`
`neighbourhood`	The neighbourhood to use. Should be intialized with the same regions. TYPE: `H3Neighbourhood`
`learning_rate`	Learning rate. Defaults to 0.001. TYPE: `float` DEFAULT: `0.001`
`trainer_kwargs`	Trainer kwargs. This is where the number of epochs can be set. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

Source code in srai/embedders/geovex/embedder.py

def fit(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: H3Neighbourhood,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> None:
    """
    Fit the model to the data.

    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.
        neighbourhood (H3Neighbourhood): The neighbourhood to use.
            Should be intialized with the same regions.
        learning_rate (float, optional): Learning rate. Defaults to 0.001.
        trainer_kwargs (Optional[Dict[str, Any]], optional): Trainer kwargs.
            This is where the number of epochs can be set. Defaults to None.
    """
    import pytorch_lightning as pl

    trainer_kwargs = self._prepare_trainer_kwargs(trainer_kwargs)
    counts_df, dataloader, dataset = self._prepare_dataset(  # type: ignore
        regions_gdf,
        features_gdf,
        joint_gdf,
        neighbourhood,
        self._batch_size,
        shuffle=True,
    )

    self._prepare_model(counts_df, learning_rate)

    trainer = pl.Trainer(**trainer_kwargs)
    trainer.fit(self._model, dataloader)
    self._is_fitted = True
    self._dataset = dataset

fit_transform ¶

fit_transform(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: H3Neighbourhood,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> pd.DataFrame

Fit the model to the data and create region embeddings.

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`
`neighbourhood`	The neighbourhood to use. Should be intialized with the same regions. TYPE: `H3Neighbourhood`
`negative_sample_k_distance`	Distance of negative samples. Defaults to 2. TYPE: `int`
`learning_rate`	Learning rate. Defaults to 0.001. TYPE: `float` DEFAULT: `0.001`
`trainer_kwargs`	Trainer kwargs. This is where the number of epochs can be set. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

Source code in srai/embedders/geovex/embedder.py

def fit_transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: H3Neighbourhood,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> pd.DataFrame:
    """
    Fit the model to the data and create region embeddings.

    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.
        neighbourhood (H3Neighbourhood): The neighbourhood to use.
            Should be intialized with the same regions.
        negative_sample_k_distance (int, optional): Distance of negative samples. Defaults to 2.
        learning_rate (float, optional): Learning rate. Defaults to 0.001.
        trainer_kwargs (Optional[Dict[str, Any]], optional): Trainer kwargs. This is where the
            number of epochs can be set. Defaults to None.
    """
    self.fit(
        regions_gdf=regions_gdf,
        features_gdf=features_gdf,
        joint_gdf=joint_gdf,
        neighbourhood=neighbourhood,
        learning_rate=learning_rate,
        trainer_kwargs=trainer_kwargs,
    )
    assert self._dataset is not None  # for mypy
    return self._transform(dataset=self._dataset)

load ¶

classmethod

load(path: Union[Path, str]) -> GeoVexEmbedder

Load the model from a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Union[Path, str]`
`model_module`	Model class. TYPE: `type[ModelT]`

RETURNS	DESCRIPTION
`GeoVexEmbedder`	GeoVexEmbedder object. TYPE: `GeoVexEmbedder`

Source code in srai/embedders/geovex/embedder.py

@classmethod
def load(cls, path: Union[Path, str]) -> "GeoVexEmbedder":
    """
    Load the model from a directory.

    Args:
        path (Union[Path, str]): Path to the directory.
        model_module (type[ModelT]): Model class.

    Returns:
        GeoVexEmbedder: GeoVexEmbedder object.
    """
    return cls._load(path, GeoVexModel)

save ¶

save(path: Union[str, Any]) -> None

Save the model to a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Union[str, Any]`

Source code in srai/embedders/geovex/embedder.py

def save(self, path: Union[str, Any]) -> None:
    """
    Save the model to a directory.

    Args:
        path (Union[str, Any]): Path to the directory.
    """
    # embedder_config must match the constructor signature:
    # target_features: Union[list[str], OsmTagsFilter, GroupedOsmTagsFilter],
    # batch_size: Optional[int] = 32,
    # neighbourhood_radius: int = 4,
    # convolutional_layers: int = 2,
    # embedding_size: int = 32,
    # convolutional_layer_size: int = 256,
    embedder_config = {
        "target_features": self.expected_output_features.to_json(orient="records"),
        "batch_size": self._batch_size,
        "neighbourhood_radius": self._r,
        "convolutional_layers": self._convolutional_layers,
        "embedding_size": self._embedding_size,
        "convolutional_layer_size": self._convolutional_layer_size,
    }
    self._save(path, embedder_config)

transform ¶

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

Create region embeddings.

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: Region embeddings.

Source code in srai/embedders/geovex/embedder.py

def transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> pd.DataFrame:
    """
    Create region embeddings.

    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: Region embeddings.
    """
    self._check_is_fitted()

    neighbourhood = H3Neighbourhood(
        regions_gdf=regions_gdf,
    )

    _, dataloader, self._dataset = self._prepare_dataset(
        regions_gdf,
        features_gdf,
        joint_gdf,
        neighbourhood,
        self._batch_size,
        shuffle=False,
    )

    return self._transform(dataset=self._dataset, dataloader=dataloader)

Hex2VecEmbedder ¶

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

Bases: CountEmbedder

Hex2Vec Embedder.

PARAMETER	DESCRIPTION
`encoder_sizes`	Sizes of the encoder layers. The input layer size shouldn't be included - it's inferred from the data. The last element is the embedding size. Defaults to [150, 75, 50]. TYPE: `List[int]` DEFAULT: `None`
`count_subcategories`	Whether to count all subcategories individually or count features only on the highest level based on features column name. Defaults to False. TYPE: `bool` DEFAULT: `True`

Source code in srai/embedders/hex2vec/embedder.py

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

    Args:
        encoder_sizes (List[int], optional): Sizes of the encoder layers.
            The input layer size shouldn't be included - it's inferred from the data.
            The last element is the embedding size. Defaults to [150, 75, 50].
        expected_output_features
            (Union[List[str], OsmTagsFilter, GroupedOsmTagsFilter], optional):
            List of expected output features. 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 False.
    """
    super().__init__(
        expected_output_features=expected_output_features,
        count_subcategories=count_subcategories,
    )
    import_optional_dependencies(
        dependency_group="torch", modules=["torch", "pytorch_lightning"]
    )
    if encoder_sizes is None:
        encoder_sizes = Hex2VecEmbedder.DEFAULT_ENCODER_SIZES
    self._assert_encoder_sizes_correct(encoder_sizes)
    self._encoder_sizes = encoder_sizes
    self._model: Optional[Hex2VecModel] = None
    self._is_fitted = False

fit ¶

fit(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: Neighbourhood[T],
    negative_sample_k_distance: int = 2,
    batch_size: int = 32,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> None

Fit the model to the data.

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`
`neighbourhood`	The neighbourhood to use. Should be intialized with the same regions. TYPE: `Neighbourhood[T]`
`negative_sample_k_distance`	When sampling negative samples, sample from a distance > k. Defaults to 2. TYPE: `int` DEFAULT: `2`
`batch_size`	Batch size. Defaults to 32. TYPE: `int` DEFAULT: `32`
`learning_rate`	Learning rate. Defaults to 0.001. TYPE: `float` DEFAULT: `0.001`
`trainer_kwargs`	Trainer kwargs. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

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 negative_sample_k_distance < 2.

Source code in srai/embedders/hex2vec/embedder.py

def fit(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: Neighbourhood[T],
    negative_sample_k_distance: int = 2,
    batch_size: int = 32,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> None:
    """
    Fit the model to the data.

    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.
        neighbourhood (Neighbourhood[T]): The neighbourhood to use.
            Should be intialized with the same regions.
        negative_sample_k_distance (int, optional): When sampling negative samples,
            sample from a distance > k. Defaults to 2.
        batch_size (int, optional): Batch size. Defaults to 32.
        learning_rate (float, optional): Learning rate. Defaults to 0.001.
        trainer_kwargs (Optional[Dict[str, Any]], optional): Trainer kwargs. Defaults to None.

    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 negative_sample_k_distance < 2.
    """
    import pytorch_lightning as pl
    from torch.utils.data import DataLoader

    trainer_kwargs = self._prepare_trainer_kwargs(trainer_kwargs)

    counts_df = self._get_raw_counts(regions_gdf, features_gdf, joint_gdf)

    if self.expected_output_features is None:  # type: ignore[has-type]
        self.expected_output_features = pd.Series(counts_df.columns)

    num_features = len(self.expected_output_features)  # type: ignore[arg-type]
    self._model = Hex2VecModel(
        layer_sizes=[num_features, *self._encoder_sizes], learning_rate=learning_rate
    )
    dataset = NeighbourDataset(counts_df, neighbourhood, negative_sample_k_distance)
    dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

    trainer = pl.Trainer(**trainer_kwargs)
    trainer.fit(self._model, dataloader)
    self._is_fitted = True

fit_transform ¶

fit_transform(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: Neighbourhood[T],
    negative_sample_k_distance: int = 2,
    batch_size: int = 32,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> pd.DataFrame

Fit the model to the data and return the embeddings.

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`
`neighbourhood`	The neighbourhood to use. Should be intialized with the same regions. TYPE: `Neighbourhood[T]`
`negative_sample_k_distance`	When sampling negative samples, sample from a distance > k. Defaults to 2. TYPE: `int` DEFAULT: `2`
`batch_size`	Batch size. Defaults to 32. TYPE: `int` DEFAULT: `32`
`learning_rate`	Learning rate. Defaults to 0.001. TYPE: `float` DEFAULT: `0.001`
`trainer_kwargs`	Trainer kwargs. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	pd.DataFrame: Region embeddings.

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 negative_sample_k_distance < 2.

Source code in srai/embedders/hex2vec/embedder.py

def fit_transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    neighbourhood: Neighbourhood[T],
    negative_sample_k_distance: int = 2,
    batch_size: int = 32,
    learning_rate: float = 0.001,
    trainer_kwargs: Optional[dict[str, Any]] = None,
) -> pd.DataFrame:
    """
    Fit the model to the data and return the embeddings.

    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.
        neighbourhood (Neighbourhood[T]): The neighbourhood to use.
            Should be intialized with the same regions.
        negative_sample_k_distance (int, optional): When sampling negative samples,
            sample from a distance > k. Defaults to 2.
        batch_size (int, optional): Batch size. Defaults to 32.
        learning_rate (float, optional): Learning rate. Defaults to 0.001.
        trainer_kwargs (Optional[Dict[str, Any]], optional): Trainer kwargs. Defaults to None.

    Returns:
        pd.DataFrame: Region embeddings.

    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 negative_sample_k_distance < 2.
    """
    self.fit(
        regions_gdf,
        features_gdf,
        joint_gdf,
        neighbourhood,
        negative_sample_k_distance,
        batch_size,
        learning_rate,
        trainer_kwargs,
    )
    return self.transform(regions_gdf, features_gdf, joint_gdf)

load ¶

classmethod

load(path: Union[Path, str]) -> Hex2VecEmbedder

Load the model from a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path`

RETURNS	DESCRIPTION
`Hex2VecEmbedder`	The loaded embedder. TYPE: `Hex2VecEmbedder`

Source code in srai/embedders/hex2vec/embedder.py

@classmethod
def load(cls, path: Union[Path, str]) -> "Hex2VecEmbedder":
    """
    Load the model from a directory.

    Args:
        path (Path): Path to the directory.

    Returns:
        Hex2VecEmbedder: The loaded embedder.
    """
    return cls._load(path, Hex2VecModel)

save ¶

save(path: Union[Path, str]) -> None

Save the model to a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path`

Source code in srai/embedders/hex2vec/embedder.py

def save(self, path: Union[Path, str]) -> None:
    """
    Save the model to a directory.

    Args:
        path (Path): Path to the directory.
    """
    embedder_config = {
        "encoder_sizes": self._encoder_sizes,
        "expected_output_features": (
            self.expected_output_features.tolist()
            if self.expected_output_features is not None
            else None
        ),
    }
    self._save(path, embedder_config)

transform ¶

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

Create region embeddings.

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 and geometry index 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.

Source code in srai/embedders/hex2vec/embedder.py

def transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> pd.DataFrame:
    """
    Create region embeddings.

    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 and geometry index 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.
    """
    import torch

    self._check_is_fitted()
    counts_df = self._get_raw_counts(regions_gdf, features_gdf, joint_gdf)
    counts_tensor = torch.from_numpy(counts_df.values)
    embeddings = self._model(counts_tensor).detach().numpy()  # type: ignore
    return pd.DataFrame(embeddings, index=counts_df.index)

Highway2VecEmbedder ¶

Highway2VecEmbedder(hidden_size: int = 64, embedding_size: int = 30)

Bases: Embedder

Highway2Vec Embedder.

PARAMETER	DESCRIPTION
`hidden_size`	Hidden size in encoder and decoder. Defaults to 64. TYPE: `int` DEFAULT: `64`
`embedding_size`	Embedding size. Defaults to 30. TYPE: `int` DEFAULT: `30`

Source code in srai/embedders/highway2vec/embedder.py

def __init__(self, hidden_size: int = 64, embedding_size: int = 30) -> None:
    """
    Init Highway2Vec Embedder.

    Args:
        hidden_size (int, optional): Hidden size in encoder and decoder. Defaults to 64.
        embedding_size (int, optional): Embedding size. Defaults to 30.
    """
    import_optional_dependencies(
        dependency_group="torch", modules=["torch", "pytorch_lightning"]
    )

    self._model: Optional[Highway2VecModel] = None
    self._hidden_size = hidden_size
    self._embedding_size = embedding_size
    self._is_fitted = False

fit ¶

fit(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    trainer_kwargs: Optional[dict[str, Any]] = None,
    dataloader_kwargs: Optional[dict[str, Any]] = None,
) -> None

Fit the model to the data.

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`
`trainer_kwargs`	Trainer kwargs. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`dataloader_kwargs`	Dataloader kwargs. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

RAISES	DESCRIPTION
`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.

Source code in srai/embedders/highway2vec/embedder.py

def fit(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    trainer_kwargs: Optional[dict[str, Any]] = None,
    dataloader_kwargs: Optional[dict[str, Any]] = None,
) -> None:
    """
    Fit the model to the data.

    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.
        trainer_kwargs (Optional[Dict[str, Any]], optional): Trainer kwargs. Defaults to None.
        dataloader_kwargs (Optional[Dict[str, Any]], optional): Dataloader kwargs.
            Defaults to None.

    Raises:
        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.
    """
    import pytorch_lightning as pl
    import torch
    from torch.utils.data import DataLoader

    self._validate_indexes(regions_gdf, features_gdf, joint_gdf)
    features_df = self._remove_geometry_if_present(features_gdf)

    num_features = len(features_df.columns)
    self._model = Highway2VecModel(
        n_features=num_features, n_hidden=self._hidden_size, n_embed=self._embedding_size
    )

    dataloader_kwargs = dataloader_kwargs or {}
    if "batch_size" not in dataloader_kwargs:
        dataloader_kwargs["batch_size"] = 128

    dataloader = DataLoader(torch.Tensor(features_df.values), **dataloader_kwargs)

    trainer_kwargs = trainer_kwargs or {}
    if "max_epochs" not in trainer_kwargs:
        trainer_kwargs["max_epochs"] = 10

    trainer = pl.Trainer(**trainer_kwargs)
    trainer.fit(self._model, dataloader)
    self._is_fitted = True

fit_transform ¶

fit_transform(
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    trainer_kwargs: Optional[dict[str, Any]] = None,
    dataloader_kwargs: Optional[dict[str, Any]] = None,
) -> pd.DataFrame

Fit the model to the data and return the embeddings.

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`
`trainer_kwargs`	Trainer kwargs. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`dataloader_kwargs`	Dataloader kwargs. Defaults to None. TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`DataFrame`	pd.DataFrame: Region embeddings.

RAISES	DESCRIPTION
`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.

Source code in srai/embedders/highway2vec/embedder.py

def fit_transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
    trainer_kwargs: Optional[dict[str, Any]] = None,
    dataloader_kwargs: Optional[dict[str, Any]] = None,
) -> pd.DataFrame:
    """
    Fit the model to the data and return the embeddings.

    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.
        trainer_kwargs (Optional[Dict[str, Any]], optional): Trainer kwargs. Defaults to None.
        dataloader_kwargs (Optional[Dict[str, Any]], optional): Dataloader kwargs.
            Defaults to None.

    Returns:
        pd.DataFrame: Region embeddings.

    Raises:
        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.
    """
    self.fit(regions_gdf, features_gdf, joint_gdf, trainer_kwargs, dataloader_kwargs)
    return self.transform(regions_gdf, features_gdf, joint_gdf)

load ¶

classmethod

load(path: Union[Path, str]) -> Highway2VecEmbedder

Load the model from a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path`

RETURNS	DESCRIPTION
`Hex2VecEmbedder`	The loaded embedder. TYPE: `Highway2VecEmbedder`

Source code in srai/embedders/highway2vec/embedder.py

@classmethod
def load(cls, path: Union[Path, str]) -> "Highway2VecEmbedder":
    """
    Load the model from a directory.

    Args:
        path (Path): Path to the directory.

    Returns:
        Hex2VecEmbedder: The loaded embedder.
    """
    return cls._load(path, Highway2VecModel)

save ¶

save(path: Union[Path, str]) -> None

Save the model to a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path`

Source code in srai/embedders/highway2vec/embedder.py

def save(self, path: Union[Path, str]) -> None:
    """
    Save the model to a directory.

    Args:
        path (Path): Path to the directory.
    """
    embedder_config = {"hidden_size": self._hidden_size, "embedding_size": self._embedding_size}
    self._save(path, embedder_config)

transform ¶

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

Embed regions using features.

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 and geometry index for each region in regions_gdf.

RAISES	DESCRIPTION
`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.

Source code in srai/embedders/highway2vec/embedder.py

def transform(
    self,
    regions_gdf: gpd.GeoDataFrame,
    features_gdf: gpd.GeoDataFrame,
    joint_gdf: gpd.GeoDataFrame,
) -> pd.DataFrame:  # pragma: no cover
    """
    Embed regions using features.

    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 and geometry index for each region in regions_gdf.

    Raises:
        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.
    """
    import torch

    self._validate_indexes(regions_gdf, features_gdf, joint_gdf)
    self._check_is_fitted()
    features_df = self._remove_geometry_if_present(features_gdf)

    self._model.eval()  # type: ignore
    embeddings = self._model(torch.Tensor(features_df.values)).detach().numpy()  # type: ignore
    embeddings_df = pd.DataFrame(embeddings, index=features_df.index)
    embeddings_joint = joint_gdf.join(embeddings_df)
    embeddings_aggregated = embeddings_joint.groupby(level=[0]).mean()

    return embeddings_aggregated

Model ¶

Bases: LightningModule

Class for model based on LightningModule.

get_config ¶

get_config() -> dict[str, Any]

Get model config.

Source code in srai/embedders/_base.py

def get_config(self) -> dict[str, Any]:
    """Get model config."""
    model_config = {
        k: v
        for k, v in vars(self).items()
        if k[0] != "_"
        and k
        not in (
            "training",
            "prepare_data_per_node",
            "allow_zero_length_dataloader_with_multiple_devices",
        )
    }

    return model_config

load ¶

classmethod

load(path: Union[Path, str], **kwargs: Any) -> Model

Load model from a file.

PARAMETER	DESCRIPTION
`path`	Path to the file. TYPE: `Union[Path, str]`
`**kwargs`	Additional kwargs to pass to the model constructor. TYPE: `dict` DEFAULT: `{}`

Source code in srai/embedders/_base.py

@classmethod
def load(cls, path: Union[Path, str], **kwargs: Any) -> "Model":
    """
    Load model from a file.

    Args:
        path (Union[Path, str]): Path to the file.
        **kwargs (dict): Additional kwargs to pass to the model constructor.
    """
    import torch

    if isinstance(path, str):
        path = Path(path)

    model = cls(**kwargs)
    model.load_state_dict(torch.load(path))
    return model

save ¶

save(path: Union[Path, str]) -> None

Save the model to a directory.

PARAMETER	DESCRIPTION
`path`	Path to the directory. TYPE: `Path`

Source code in srai/embedders/_base.py

def save(self, path: Union[Path, str]) -> None:
    """
    Save the model to a directory.

    Args:
        path (Path): Path to the directory.
    """
    import torch

    torch.save(self.state_dict(), path)

Index

embedders ¶

ContextualCountEmbedder ¶

transform ¶

CountEmbedder ¶

transform ¶

Embedder ¶

transform ¶

GTFS2VecEmbedder ¶

fit ¶

fit_transform ¶

load ¶

save ¶

transform ¶

GeoVexEmbedder ¶

invalid_cells property ¶

fit ¶

fit_transform ¶

load ¶

save ¶

transform ¶

Hex2VecEmbedder ¶

fit ¶

fit_transform ¶

load ¶

save ¶

transform ¶

Highway2VecEmbedder ¶

fit ¶

fit_transform ¶

load ¶

save ¶

transform ¶

Model ¶

get_config ¶

load ¶

save ¶

invalid_cells `property` ¶