Skip to content

Data Normalization

Normalization improves neural network training by ensuring features are well-scaled, preventing issues like exploding gradients and slow convergence. In power grids, where variables like voltage and power span wide ranges, normalization is essential. The gridfm-graphkit package offers four methods:

Each of these strategies implements a unified interface and can be used interchangeably depending on the learning task and data characteristics.

Users can create their own custom normalizers by extending the base Normalizer class to suit specific needs.

Available Normalizers

Normalizer

Bases: ABC

Abstract base class for all normalization strategies.

Source code in gridfm_graphkit/datasets/normalizers.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
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
class Normalizer(ABC):
    """
    Abstract base class for all normalization strategies.
    """

    @abstractmethod
    def fit(self, data: torch.Tensor) -> dict:
        """
        Fit normalization parameters from data.

        Args:
            data: Input tensor.

        Returns:
            Dictionary of computed parameters.
        """

    @abstractmethod
    def fit_from_dict(self, params: dict):
        """
        Set parameters from a precomputed dictionary.

        Args:
            params: Dictionary of parameters.
        """

    @abstractmethod
    def transform(self, data: torch.Tensor) -> torch.Tensor:
        """
        Normalize the input data.

        Args:
            data: Input tensor.

        Returns:
            Normalized tensor.
        """

    @abstractmethod
    def inverse_transform(self, normalized_data: torch.Tensor) -> torch.Tensor:
        """
        Undo normalization.

        Args:
            normalized_data: Normalized tensor.

        Returns:
            Original tensor.
        """

    @abstractmethod
    def get_stats(self) -> dict:
        """
        Return the stored normalization statistics for logging/inspection.
        """
fit(data) abstractmethod

Fit normalization parameters from data.

Parameters:

Name Type Description Default
data Tensor

Input tensor.

 required

Returns:

Type Description
dict

Dictionary of computed parameters.
Source code in gridfm_graphkit/datasets/normalizers.py 
12
13
14
15
16
17
18
19
20
21
22
@abstractmethod
def fit(self, data: torch.Tensor) -> dict:
    """
    Fit normalization parameters from data.

    Args:
        data: Input tensor.

    Returns:
        Dictionary of computed parameters.
    """
fit_from_dict(params) abstractmethod

Set parameters from a precomputed dictionary.

Parameters:

Name Type Description Default
params dict

Dictionary of parameters.

 required
Source code in gridfm_graphkit/datasets/normalizers.py 
24
25
26
27
28
29
30
31
@abstractmethod
def fit_from_dict(self, params: dict):
    """
    Set parameters from a precomputed dictionary.

    Args:
        params: Dictionary of parameters.
    """
get_stats() abstractmethod

Return the stored normalization statistics for logging/inspection.

Source code in gridfm_graphkit/datasets/normalizers.py 
57
58
59
60
61
@abstractmethod
def get_stats(self) -> dict:
    """
    Return the stored normalization statistics for logging/inspection.
    """
inverse_transform(normalized_data) abstractmethod

Undo normalization.

Parameters:

Name Type Description Default
normalized_data Tensor

Normalized tensor.

 required

Returns:

Type Description
Tensor

Original tensor.
Source code in gridfm_graphkit/datasets/normalizers.py 
45
46
47
48
49
50
51
52
53
54
55
@abstractmethod
def inverse_transform(self, normalized_data: torch.Tensor) -> torch.Tensor:
    """
    Undo normalization.

    Args:
        normalized_data: Normalized tensor.

    Returns:
        Original tensor.
    """
transform(data) abstractmethod

Normalize the input data.

Parameters:

Name Type Description Default
data Tensor

Input tensor.

 required

Returns:

Type Description
Tensor

Normalized tensor.
Source code in gridfm_graphkit/datasets/normalizers.py 
33
34
35
36
37
38
39
40
41
42
43
@abstractmethod
def transform(self, data: torch.Tensor) -> torch.Tensor:
    """
    Normalize the input data.

    Args:
        data: Input tensor.

    Returns:
        Normalized tensor.
    """

MinMaxNormalizer

Bases: Normalizer

Scales each feature to the [0, 1] range.

Parameters:

Name Type Description Default
node_data bool

Whether data is node-level or edge-level

 required
args NestedNamespace

Parameters

 required
Source code in gridfm_graphkit/datasets/normalizers.py 
 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
@NORMALIZERS_REGISTRY.register("minmax")
class MinMaxNormalizer(Normalizer):
    """
    Scales each feature to the [0, 1] range.

    Args:
        node_data (bool): Whether data is node-level or edge-level
        args (NestedNamespace): Parameters

    """

    def __init__(self, node_data: bool, args):
        self.min_val = None
        self.max_val = None

    def to(self, device):
        self.min_val = self.min_val.to(device)
        self.max_val = self.max_val.to(device)

    def fit(self, data: torch.Tensor) -> dict:
        self.min_val, _ = data.min(axis=0)
        self.max_val, _ = data.max(axis=0)

        return {"min_value": self.min_val, "max_value": self.max_val}

    def fit_from_dict(self, params: dict):
        if self.min_val is None:
            self.min_val = params.get("min_value")
        if self.max_val is None:
            self.max_val = params.get("max_value")

    def transform(self, data: torch.Tensor) -> torch.Tensor:
        if self.min_val is None or self.max_val is None:
            raise ValueError("fit must be called before transform.")

        diff = self.max_val - self.min_val
        diff[diff == 0] = 1  # Avoid division by zero for features with zero range
        return (data - self.min_val) / diff

    def inverse_transform(self, normalized_data: torch.Tensor) -> torch.Tensor:
        if self.min_val is None or self.max_val is None:
            raise ValueError("fit must be called before inverse_transform.")

        diff = self.max_val - self.min_val
        diff[diff == 0] = 1
        return (normalized_data * diff) + self.min_val

    def get_stats(self) -> dict:
        return {
            "min_value": self.min_val.tolist() if self.min_val is not None else None,
            "max_value": self.max_val.tolist() if self.max_val is not None else None,
        }

Standardizer

Bases: Normalizer

Standardizes each feature to zero mean and unit variance.

Parameters:

Name Type Description Default
node_data bool

Whether data is node-level or edge-level

 required
args NestedNamespace

Parameters

 required
Source code in gridfm_graphkit/datasets/normalizers.py 
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
160
161
162
163
164
165
166
167
168
169
@NORMALIZERS_REGISTRY.register("standard")
class Standardizer(Normalizer):
    """
    Standardizes each feature to zero mean and unit variance.

    Args:
        node_data (bool): Whether data is node-level or edge-level
        args (NestedNamespace): Parameters

    """

    def __init__(self, node_data: bool, args):
        self.mean = None
        self.std = None

    def to(self, device):
        self.mean = self.mean.to(device)
        self.std = self.std.to(device)

    def fit(self, data: torch.Tensor) -> dict:
        self.mean = data.mean(axis=0)
        self.std = data.std(axis=0)

        return {"mean_value": self.mean, "std_value": self.std}

    def fit_from_dict(self, params: dict):
        if self.mean is None:
            self.mean = params.get("mean_value")
        if self.std is None:
            self.std = params.get("std_value")

    def transform(self, data: torch.Tensor) -> torch.Tensor:
        if self.mean is None or self.std is None:
            raise ValueError("fit must be called before transform.")

        std = self.std.clone()
        std[std == 0] = 1  # Avoid division by zero for features with zero std
        return (data - self.mean) / std

    def inverse_transform(self, normalized_data: torch.Tensor) -> torch.Tensor:
        if self.mean is None or self.std is None:
            raise ValueError("fit must be called before inverse_transform.")

        std = self.std.clone()
        std[std == 0] = 1
        return (normalized_data * std) + self.mean

    def get_stats(self) -> dict:
        return {
            "mean": self.mean.tolist() if self.mean is not None else None,
            "std": self.std.tolist() if self.std is not None else None,
        }

BaseMVANormalizer

Bases: Normalizer

In power systems, a suitable normalization strategy must preserve the physical properties of the system. A known method is the conversion to the per-unit (p.u.) system, which expresses electrical quantities such as voltage, current, power, and impedance as fractions of predefined base values. These base values are usually chosen based on system parameters, such as rated voltage. The per-unit conversion ensures that power system equations remain scale-invariant, preserving fundamental physical relationships.

Source code in gridfm_graphkit/datasets/normalizers.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@NORMALIZERS_REGISTRY.register("baseMVAnorm")
class BaseMVANormalizer(Normalizer):
    """
    In power systems, a suitable normalization strategy must preserve the physical properties of
    the system. A known method is the conversion to the per-unit (p.u.) system, which expresses
    electrical quantities such as voltage, current, power, and impedance as fractions of predefined
    base values. These base values are usually chosen based on system parameters, such as rated
    voltage. The per-unit conversion ensures that power system equations remain scale-invariant,
    preserving fundamental physical relationships.
    """

    def __init__(self, node_data: bool, args):
        """
        Args:
            node_data: Whether data is node-level or edge-level
            args (NestedNamespace): Parameters

        Attributes:
            baseMVA (float): baseMVA found in casefile. From ``args.data.baseMVA``.
        """
        self.node_data = node_data
        self.baseMVA_orig = getattr(args.data, "baseMVA", 100)
        self.baseMVA = None

    def to(self, device):
        pass

    def fit(self, data: torch.Tensor, baseMVA: float = None) -> dict:
        if self.node_data:
            self.baseMVA = data[:, [PD, QD, PG, QG]].max()
        else:
            self.baseMVA = baseMVA

        return {"baseMVA_orig": self.baseMVA_orig, "baseMVA": self.baseMVA}

    def fit_from_dict(self, params: dict):
        if self.baseMVA is None:
            self.baseMVA = params.get("baseMVA")
        if self.baseMVA_orig is None:
            self.baseMVA_orig = params.get("baseMVA_orig")

    def transform(self, data: torch.Tensor) -> torch.Tensor:
        if self.baseMVA is None:
            raise ValueError("BaseMVA is not specified")

        if self.baseMVA == 0:
            raise ZeroDivisionError("BaseMVA is 0.")

        if self.node_data:
            data[:, PD] = data[:, PD] / self.baseMVA
            data[:, QD] = data[:, QD] / self.baseMVA
            data[:, PG] = data[:, PG] / self.baseMVA
            data[:, QG] = data[:, QG] / self.baseMVA
            data[:, VA] = data[:, VA] * torch.pi / 180.0
        else:
            data = data * self.baseMVA_orig / self.baseMVA

        return data

    def inverse_transform(self, normalized_data: torch.Tensor) -> torch.Tensor:
        if self.baseMVA is None:
            raise ValueError("fit must be called before inverse_transform.")

        if self.node_data:
            normalized_data[:, PD] = normalized_data[:, PD] * self.baseMVA
            normalized_data[:, QD] = normalized_data[:, QD] * self.baseMVA
            normalized_data[:, PG] = normalized_data[:, PG] * self.baseMVA
            normalized_data[:, QG] = normalized_data[:, QG] * self.baseMVA
            normalized_data[:, VA] = normalized_data[:, VA] * 180.0 / torch.pi
        else:
            normalized_data = normalized_data * self.baseMVA / self.baseMVA_orig

        return normalized_data

    def get_stats(self) -> dict:
        return {
            "baseMVA": self.baseMVA,
            "baseMVA_orig": self.baseMVA_orig,
        }
__init__(node_data, args)

Parameters:

Name Type Description Default
node_data bool

Whether data is node-level or edge-level

 required
args NestedNamespace

Parameters

 required

Attributes:

Name Type Description
baseMVA float

baseMVA found in casefile. From args.data.baseMVA.
Source code in gridfm_graphkit/datasets/normalizers.py 
183
184
185
186
187
188
189
190
191
192
193
194
def __init__(self, node_data: bool, args):
    """
    Args:
        node_data: Whether data is node-level or edge-level
        args (NestedNamespace): Parameters

    Attributes:
        baseMVA (float): baseMVA found in casefile. From ``args.data.baseMVA``.
    """
    self.node_data = node_data
    self.baseMVA_orig = getattr(args.data, "baseMVA", 100)
    self.baseMVA = None

IdentityNormalizer

Bases: Normalizer

No normalization: returns data unchanged.

Parameters:

Name Type Description Default
node_data bool

Whether data is node-level or edge-level

 required
args NestedNamespace

Parameters

 required
Source code in gridfm_graphkit/datasets/normalizers.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@NORMALIZERS_REGISTRY.register("identity")
class IdentityNormalizer(Normalizer):
    """
    No normalization: returns data unchanged.

    Args:
            node_data: Whether data is node-level or edge-level
            args (NestedNamespace): Parameters
    """

    def __init__(self, node_data: bool, args):
        pass

    def fit(self, data: torch.Tensor) -> dict:
        return {}

    def fit_from_dict(self, params: dict):
        pass

    def transform(self, data: torch.Tensor) -> torch.Tensor:
        return data

    def inverse_transform(self, normalized_data: torch.Tensor) -> torch.Tensor:
        return normalized_data

    def get_stats(self) -> dict:
        return {"note": "No normalization applied."}

Usage Workflow

Example:

from gridfm_graphkit.datasets.normalizers import MinMaxNormalizer
import torch

data = torch.randn(100, 5)  # Example tensor

normalizer = MinMaxNormalizer(node_data=True,args=None)
params = normalizer.fit(data)
normalized = normalizer.transform(data)
restored = normalizer.inverse_transform(normalized)