Skip to content

Highway2VecEmbedder

Bases: Embedder

Highway2Vec Embedder.

Source code in srai/embedders/highway2vec/embedder.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class Highway2VecEmbedder(Embedder):
    """Highway2Vec Embedder."""

    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

    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

    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

    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)

    def _check_is_fitted(self) -> None:
        if not self._is_fitted or self._model is None:
            raise ModelNotFitException("Model not fitted. Call fit() or fit_transform() first.")

__init__ 

__init__(hidden_size: int = 64, embedding_size: int = 30) -> None

Init 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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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: gpd.GeoDataFrame

features_gdf

Feature indexes, geometries and feature values.

TYPE: gpd.GeoDataFrame

joint_gdf

Joiner result with region-feature multi-index.

TYPE: gpd.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 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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: gpd.GeoDataFrame

features_gdf

Feature indexes, geometries and feature values.

TYPE: gpd.GeoDataFrame

joint_gdf

Joiner result with region-feature multi-index.

TYPE: gpd.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
pd.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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
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)

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: gpd.GeoDataFrame

features_gdf

Feature indexes, geometries and feature values.

TYPE: gpd.GeoDataFrame

joint_gdf

Joiner result with region-feature multi-index.

TYPE: gpd.GeoDataFrame

RETURNS DESCRIPTION
pd.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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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