• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

tonegas / nnodely / 14268084640

04 Apr 2025 02:51PM UTC coverage: 97.035% (+0.04%) from 96.995%
14268084640

push

github

web-flow
Merge pull request #82 from tonegas/develop

Added some new features:

1. the import from pandas dataframe with resample feature
2. new example files for each layers
3. categorical loss

407 of 430 new or added lines in 20 files covered. (94.65%)

7 existing lines in 2 files now uncovered.

11453 of 11803 relevant lines covered (97.03%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.04
/nnodely/fuzzify.py
1
import inspect, copy, textwrap, torch
1✔
2

3
import numpy as np
1✔
4
import torch.nn as nn
1✔
5

6
from collections.abc import Callable
1✔
7

8
from nnodely.relation import NeuObj, Stream
1✔
9
from nnodely.model import Model
1✔
10
from nnodely.utils import check, merge, enforce_types
1✔
11

12
from nnodely.logger import logging, nnLogger
1✔
13
log = nnLogger(__name__, logging.CRITICAL)
1✔
14

15
fuzzify_relation_name = 'Fuzzify'
1✔
16

17

18
class Fuzzify(NeuObj):
1✔
19
    """
20
    Represents a Fuzzify relation in the neural network model.
21

22
    Parameters
23
    ----------
24
    output_dimension : int, optional
25
        The output dimension of the Fuzzify relation. If provided, `range` must also be provided and `centers` must be None.
26
    range : list, optional
27
        A list containing the start and end values for the range. Required if `output_dimension` is provided.
28
    centers : list, optional
29
        A list of center values for the fuzzy functions. Required if `output_dimension` is None.
30
        The `output_dimension` will be inferred from the number of centers provided.
31
    functions : str, list, or Callable, optional
32
        The fuzzy functions to use. Can be a string specifying a predefined function type, a custom function, or a list of callable functions. Default is 'Triangular'.
33
    
34
    Notes
35
    -----
36
    .. note::
37
        The predefined function types are 'Triangular' and 'Rectangular'.
38
        It is also possible to pass a list of custom functions. In this case, each center will be associated with the respective function in the list.
39

40
    Attributes
41
    ----------
42
    relation_name : str
43
        The name of the relation.
44
    output_dimension : dict
45
        The output dimension of the Fuzzify relation.
46
    json : dict
47
        A dictionary containing the configuration of the Fuzzify relation.
48

49
    Examples
50
    --------
51
    .. image:: https://colab.research.google.com/assets/colab-badge.svg
52
        :target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/fuzzify.ipynb
53
        :alt: Open in Colab
54
        
55
    Example - basic usage:
56
        >>> x = Input('x')
57
        >>> fuz = Fuzzify(output_dimension=5, range=[1,5])
58
        >>> out = Output('out', fuz(x.last()))
59

60
    Example - passing the centers:
61
        >>> fuz = Fuzzify(centers=[-1,0,3,5], functions='Rectangular')
62
        >>> out = Output('out', fuz(x.last()))
63

64
    Example - using a custom function:
65
        >>> def fun(x):
66
        >>>     import torch
67
        >>>     return torch.tanh(x)
68
        >>> fuz = Fuzzify(output_dimension=11, range=[-5,5], functions=fun)
69
        >>> out = Output('out', fuz(x.last()))
70
    """
71
    @enforce_types
1✔
72
    def __init__(self, output_dimension: int | None = None,
1✔
73
                 range: list | None = None,
74
                 centers: list | None = None,
75
                 functions: str | list | Callable = 'Triangular'):
76

77
        self.relation_name = fuzzify_relation_name
1✔
78
        super().__init__('F' + fuzzify_relation_name + str(NeuObj.count))
1✔
79
        self.json['Functions'][self.name] = {}
1✔
80
        if output_dimension is not None:
1✔
81
            check(range is not None, ValueError, 'if "output_dimension" is not None, "range" must be not setted')
1✔
82
            check(centers is None, ValueError,
1✔
83
                  'if "output_dimension" and "range" are not None, then "centers" must be None')
84
            self.output_dimension = {'dim': output_dimension}
1✔
85
            interval = ((range[1] - range[0]) / (output_dimension - 1))
1✔
86
            self.json['Functions'][self.name]['centers'] = np.arange(range[0], range[1] + interval, interval).tolist()
1✔
87
        else:
88
            check(centers is not None, ValueError, 'if "output_dimension" is None and "centers" must be setted')
1✔
89
            self.output_dimension = {'dim': len(centers)}
1✔
90
            self.json['Functions'][self.name]['centers'] = np.array(centers).tolist()
1✔
91
        self.json['Functions'][self.name]['dim_out'] = copy.deepcopy(self.output_dimension)
1✔
92

93
        if type(functions) is str:
1✔
94
            self.json['Functions'][self.name]['functions'] = functions
1✔
95
            self.json['Functions'][self.name]['names'] = functions
1✔
96
        elif type(functions) is list:
1✔
97
            self.json['Functions'][self.name]['functions'] = []
1✔
98
            self.json['Functions'][self.name]['names'] = []
1✔
99
            for func in functions:
1✔
100
                code = textwrap.dedent(inspect.getsource(func)).replace('\"', '\'')
1✔
101
                self.json['Functions'][self.name]['functions'].append(code)
1✔
102
                self.json['Functions'][self.name]['names'].append(func.__name__)
1✔
103
        else:
104
            code = textwrap.dedent(inspect.getsource(functions)).replace('\"', '\'')
1✔
105
            self.json['Functions'][self.name]['functions'] = code
1✔
106
            self.json['Functions'][self.name]['names'] = functions.__name__
1✔
107

108
    @enforce_types
1✔
109
    def __call__(self, obj: Stream) -> Stream:
1✔
110
        stream_name = fuzzify_relation_name + str(Stream.count)
1✔
111
        check(type(obj) is Stream, TypeError,
1✔
112
              f"The type of {obj} is {type(obj)} and is not supported for Fuzzify operation.")
113
        check('dim' in obj.dim and obj.dim['dim'] == 1, ValueError, 'Input dimension must be scalar')
1✔
114
        output_dimension = copy.deepcopy(obj.dim)
1✔
115
        output_dimension.update(self.output_dimension)
1✔
116
        stream_json = merge(self.json, obj.json)
1✔
117
        stream_json['Relations'][stream_name] = [fuzzify_relation_name, [obj.name], self.name]
1✔
118
        return Stream(stream_name, stream_json, output_dimension)
1✔
119

120

121
def return_fuzzify(json, xlim=None, num_points=1000):
1✔
122
    if xlim is not None:
1✔
123
        x = torch.from_numpy(np.linspace(xlim[0], xlim[1], num=num_points))
×
124
    else:
125
        x = torch.from_numpy(np.linspace(json['centers'][0] - 2, json['centers'][-1] + 2, num=num_points))
1✔
126
    chan_centers = np.array(json['centers'])
1✔
127
    activ_fun = {}
1✔
128
    if isinstance(json['names'], list):
1✔
129
        n_func = len(json['names'])
1✔
130
    else:
131
        n_func = 1
1✔
132
    for i in range(len(chan_centers)):
1✔
133
        if json['functions'] == 'Triangular':
1✔
UNCOV
134
            activ_fun[i] = triangular(x, i, chan_centers).tolist()
×
135
        elif json['functions'] == 'Rectangular':
1✔
136
            activ_fun[i] = rectangular(x, i, chan_centers).tolist()
×
137
        else:
138
            if isinstance(json['names'], list):
1✔
139
                if i >= n_func:
1✔
140
                    func_idx = i - round(n_func * (i // n_func))
1✔
141
                else:
142
                    func_idx = i
1✔
143
                exec(json['functions'][func_idx], globals())
1✔
144
                function_to_call = globals()[json['names'][func_idx]]
1✔
145
            else:
146
                exec(json['functions'], globals())
1✔
147
                function_to_call = globals()[json['names']]
1✔
148
            activ_fun[i] = custom_function(function_to_call, x, i, chan_centers).tolist()
1✔
149
    return x.tolist(), activ_fun
1✔
150

151

152
def triangular(x, idx_channel, chan_centers):
1✔
153
    # Compute the number of channels
154
    num_channels = len(chan_centers)
1✔
155

156
    # First dimension of activation
157
    if idx_channel == 0:
1✔
158
        if num_channels != 1:
1✔
159
            ampl = chan_centers[1] - chan_centers[0]
1✔
160
            act_fcn = torch.minimum(torch.maximum(-(x - chan_centers[0]) / ampl + 1, torch.tensor(0.0)),
1✔
161
                                    torch.tensor(1.0))
162
        else:
163
            # In case the user only wants one channel
164
            act_fcn = 1
×
165
    elif idx_channel != 0 and idx_channel == (num_channels - 1):
1✔
166
        ampl = chan_centers[-1] - chan_centers[-2]
1✔
167
        act_fcn = torch.minimum(torch.maximum((x - chan_centers[-2]) / ampl, torch.tensor(0.0)), torch.tensor(1.0))
1✔
168
    else:
169
        ampl_1 = chan_centers[idx_channel] - chan_centers[idx_channel - 1]
1✔
170
        ampl_2 = chan_centers[idx_channel + 1] - chan_centers[idx_channel]
1✔
171
        act_fcn = torch.minimum(torch.maximum((x - chan_centers[idx_channel - 1]) / ampl_1, torch.tensor(0.0)),
1✔
172
                                torch.maximum(-(x - chan_centers[idx_channel]) / ampl_2 + 1, torch.tensor(0.0)))
173

174
    return act_fcn
1✔
175

176

177
def rectangular(x, idx_channel, chan_centers):
1✔
178
    ## compute number of channels
179
    num_channels = len(chan_centers)
1✔
180

181
    ## First dimension of activation
182
    if idx_channel == 0:
1✔
183
        if num_channels != 1:
1✔
184
            width = abs(chan_centers[idx_channel + 1] - chan_centers[idx_channel]) / 2
1✔
185
            act_fcn = torch.where(x < (chan_centers[idx_channel] + width), 1.0, 0.0)
1✔
186
        else:
187
            # In case the user only wants one channel
188
            act_fcn = 1.0
×
189
    elif idx_channel != 0 and idx_channel == (num_channels - 1):
1✔
190
        width = abs(chan_centers[idx_channel] - chan_centers[idx_channel - 1]) / 2
1✔
191
        act_fcn = torch.where(x >= chan_centers[idx_channel] - width, 1.0, 0.0)
1✔
192
    else:
193
        width_forward = abs(chan_centers[idx_channel + 1] - chan_centers[idx_channel]) / 2
1✔
194
        width_backward = abs(chan_centers[idx_channel] - chan_centers[idx_channel - 1]) / 2
1✔
195
        act_fcn = torch.where(
1✔
196
            (x >= chan_centers[idx_channel] - width_backward) & (x < chan_centers[idx_channel] + width_forward), 1.0,
197
            0.0)
198

199
    return act_fcn
1✔
200

201

202
def custom_function(func, x, idx_channel, chan_centers):
1✔
203
    act_fcn = func(x - chan_centers[idx_channel])
1✔
204
    return act_fcn
1✔
205

206

207
class Fuzzify_Layer(nn.Module):
1✔
208
    def __init__(self, params):
1✔
209
        super().__init__()
1✔
210
        self.centers = params['centers']
1✔
211
        self.function = params['functions']
1✔
212
        self.dimension = params['dim_out']['dim']
1✔
213
        self.name = params['names']
1✔
214

215
        if type(self.name) is list:
1✔
216
            self.n_func = len(self.name)
1✔
217
            for func, name in zip(self.function, self.name):
1✔
218
                ## Add the function to the globals
219
                try:
1✔
220
                    code = 'import torch\n@torch.fx.wrap\n' + func
1✔
221
                    exec(code, globals())
1✔
222
                except Exception as e:
×
223
                    check(False, RuntimeError, f"An error occurred when running the function '{name}':\n {e}")
×
224
        else:
225
            self.n_func = 1
1✔
226
            if self.name not in ['Triangular', 'Rectangular']:  ## custom function
1✔
227
                ## Add the function to the globals
228
                try:
1✔
229
                    code = 'import torch\n@torch.fx.wrap\n' + self.function
1✔
230
                    exec(code, globals())
1✔
231
                except Exception as e:
×
232
                    check(False, RuntimeError, f"An error occurred when running the function '{self.name}':\n {e}")
×
233

234
    def forward(self, x):
1✔
235
        # res = torch.empty((x.size(0), x.size(1), self.dimension), dtype=torch.float32)
236
        res = torch.zeros_like(x).repeat(1, 1, self.dimension)
1✔
237

238
        if self.function == 'Triangular':
1✔
239
            for i in range(len(self.centers)):
1✔
240
                # res[:, :, i:i+1] = triangular(x, i, self.centers)
241
                slicing(res, torch.tensor(i), triangular(x, i, self.centers))
1✔
242
        elif self.function == 'Rectangular':
1✔
243
            for i in range(len(self.centers)):
1✔
244
                # res[:, :, i:i+1] = rectangular(x, i, self.centers)
245
                slicing(res, torch.tensor(i), rectangular(x, i, self.centers))
1✔
246
        else:  ## Custom_function
247
            if self.n_func == 1:
1✔
248
                # Retrieve the function object from the globals dictionary
249
                function_to_call = globals()[self.name]
1✔
250
                for i in range(len(self.centers)):
1✔
251
                    # res[:, :, i:i+1] = custom_function(function_to_call, x, i, self.centers)
252
                    slicing(res, torch.tensor(i), custom_function(function_to_call, x, i, self.centers))
1✔
253
            else:  ## we have multiple functions
254
                for i in range(len(self.centers)):
1✔
255
                    if i >= self.n_func:
1✔
256
                        func_idx = i - round(self.n_func * (i // self.n_func))
1✔
257
                    else:
258
                        func_idx = i
1✔
259
                    function_to_call = globals()[self.name[func_idx]]
1✔
260
                    # res[:, :, i:i+1] = custom_function(function_to_call, x, i, self.centers)
261
                    slicing(res, torch.tensor(i), custom_function(function_to_call, x, i, self.centers))
1✔
262
        return res
1✔
263

264

265
@torch.fx.wrap
1✔
266
def slicing(res, i, x):
1✔
267
    res[:, :, i:i + 1] = x
1✔
268

269

270
def createFuzzify(self, *params):
1✔
271
    return Fuzzify_Layer(params[0])
1✔
272

273

274
setattr(Model, fuzzify_relation_name, createFuzzify)
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc