13504608544

Committed 24 Feb 2025 05:59PM UTC coverage: 95.261% (+0.3%) from 94.961%

Build # 13504608544

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #59 from tonegas/features/56-dynamic-parametric-function

Fix the parameter problem when defining different streams. In particular for Linear, ParametricFunction and Fir.

Run Details

567 of 582 new or added lines in 24 files covered. (97.42%)

3 existing lines in 3 files now uncovered.

10171 of 10677 relevant lines covered (95.26%)

0.95 hits per line

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

89.66

/nnodely/localmodel.py

import inspect

from collections.abc import Callable

from nnodely.relation import NeuObj, Stream
from nnodely.part import Select
from nnodely.utils import check, enforce_types

localmodel_relation_name = 'LocalModel'

class LocalModel(NeuObj):
    """
    Represents a Local Model relation in the neural network model.

    Parameters
    ----------
    input_function : Callable, optional
        A callable function to process the inputs. 
    output_function : Callable, optional
        A callable function to process the outputs. 
    pass_indexes : bool, optional
        A boolean indicating whether to pass indexes to the functions. Default is False.

    Attributes
    ----------
    relation_name : str
        The name of the relation.
    pass_indexes : bool
        A boolean indicating whether to pass indexes to the functions.
    input_function : Callable
        The function to process the inputs.
    output_function : Callable
        The function to process the outputs.

    Examples
    --------

    Example - basic usage:
        >>> x = Input('x')
        >>> activation = Fuzzify(2,[0,1],functions='Triangular')(x.last())
        >>> loc = LocalModel(input_function=Fir())
        >>> out = Output('out', loc(x.tw(1), activation))

    Example - passing a custom function:
        >>> def myFun(in1,p1,p2):
        >>>     return p1*in1+p2

        >>> x = Input('x')
        >>> activation = Fuzzify(2,[0,1],functions='Triangular')(x.last())
        >>> loc = LocalModel(input_function = lambda:ParamFun(myFun), output_function = lambda:Fir)(x.last(), activation)
        >>> out = Output('out', loc)

    Example - custom function with multiple activations:
        >>> x = Input('x')
        >>> F = Input('F')
        >>> activationA = Fuzzify(2,[0,1],functions='Triangular')(x.tw(1))
        >>> activationB = Fuzzify(2,[0,1],functions='Triangular')(F.tw(1))

        >>> def myFun(in1,p1,p2):
        >>>     return p1*in1+p2

        >>> loc = LocalModel(input_function = lambda:ParamFun(myFun), output_function = Fir(3))(x.tw(1),(activationA,activationB))
        >>> out = Output('out', loc)
    """
    @enforce_types
    def __init__(self, input_function:Callable|None = None,
                 output_function:Callable|None = None,
                 pass_indexes:bool = False):

        self.relation_name = localmodel_relation_name
        self.pass_indexes = pass_indexes
        super().__init__(localmodel_relation_name + str(NeuObj.count))
        self.json['Functions'][self.name] = {}
        if input_function is not None:
            check(callable(input_function), TypeError, 'The input_function must be callable')
        self.input_function = input_function
        if output_function is not None:
            check(callable(output_function), TypeError, 'The output_function must be callable')
        self.output_function = output_function

    @enforce_types
    def __call__(self, inputs:Stream|tuple, activations:Stream|tuple= None):
        out_sum = []
        if type(activations) is not tuple:
            activations = (activations,)
        self.___activations_matrix(activations,inputs,out_sum)

        out = out_sum[0]
        for ind in range(1,len(out_sum)):
            out = out + out_sum[ind]
        return out

    # Definisci una funzione ricorsiva per annidare i cicli for
    def ___activations_matrix(self, activations, inputs, out, idx=0, idx_list=[]):
        if idx != len(activations):
            for i in range(activations[idx].dim['dim']):
                self.___activations_matrix(activations, inputs, out, idx+1, idx_list+[i])
        else:
            if self.input_function is not None:
                if len(inspect.getfullargspec(self.input_function).args) == 0:
                    if type(inputs) is tuple:
                        out_in = self.input_function()(*inputs)
                    else:
                        out_in = self.input_function()(inputs)
                else:
                    if self.pass_indexes:
                        if type(inputs) is tuple:
                            out_in = self.input_function(idx_list)(*inputs)
                        else:
                            out_in = self.input_function(idx_list)(inputs)
                    else:
                        if type(inputs) is tuple:
                            out_in = self.input_function(*inputs)
                        else:
                            out_in = self.input_function(inputs)
            else:
                check(type(inputs) is not tuple, TypeError, 'The input cannot be a tuple without input_function')
                out_in = inputs

            act = Select(activations[0], idx_list[0])
            for ind, i  in enumerate(idx_list[1:]):
                act = act * Select(activations[ind+1], i)

            prod = out_in * act

            if self.output_function is not None:
                if len(inspect.signature(self.output_function).parameters) == 0:
                    out.append(self.output_function()(prod))
                else:
                    if self.pass_indexes:
                        out.append(self.output_function(idx_list)(prod))
                    else:
                        out.append(self.output_function(prod))
            else:
                out.append(prod)

1	import inspect	1✔
2
3	from collections.abc import Callable	1✔
4
5	from nnodely.relation import NeuObj, Stream	1✔
6	from nnodely.part import Select	1✔
7	from nnodely.utils import check, enforce_types	1✔
8
9	localmodel_relation_name = 'LocalModel'	1✔
10
11	class LocalModel(NeuObj):	1✔
12	"""
13	Represents a Local Model relation in the neural network model.
14
15	Parameters
16	----------
17	input_function : Callable, optional
18	A callable function to process the inputs.
19	output_function : Callable, optional
20	A callable function to process the outputs.
21	pass_indexes : bool, optional
22	A boolean indicating whether to pass indexes to the functions. Default is False.
23
24	Attributes
25	----------
26	relation_name : str
27	The name of the relation.
28	pass_indexes : bool
29	A boolean indicating whether to pass indexes to the functions.
30	input_function : Callable
31	The function to process the inputs.
32	output_function : Callable
33	The function to process the outputs.
34
35	Examples
36	--------
37
38	Example - basic usage:
39	>>> x = Input('x')
40	>>> activation = Fuzzify(2,[0,1],functions='Triangular')(x.last())
41	>>> loc = LocalModel(input_function=Fir())
42	>>> out = Output('out', loc(x.tw(1), activation))
43
44	Example - passing a custom function:
45	>>> def myFun(in1,p1,p2):
46	>>> return p1*in1+p2
47
48	>>> x = Input('x')
49	>>> activation = Fuzzify(2,[0,1],functions='Triangular')(x.last())
50	>>> loc = LocalModel(input_function = lambda:ParamFun(myFun), output_function = lambda:Fir)(x.last(), activation)
51	>>> out = Output('out', loc)
52
53	Example - custom function with multiple activations:
54	>>> x = Input('x')
55	>>> F = Input('F')
56	>>> activationA = Fuzzify(2,[0,1],functions='Triangular')(x.tw(1))
57	>>> activationB = Fuzzify(2,[0,1],functions='Triangular')(F.tw(1))
58
59	>>> def myFun(in1,p1,p2):
60	>>> return p1*in1+p2
61
62	>>> loc = LocalModel(input_function = lambda:ParamFun(myFun), output_function = Fir(3))(x.tw(1),(activationA,activationB))
63	>>> out = Output('out', loc)
64	"""
65	@enforce_types	1✔
66	def __init__(self, input_function:Callable\|None = None,	1✔
67	output_function:Callable\|None = None,
68	pass_indexes:bool = False):
69
70	self.relation_name = localmodel_relation_name	1✔
71	self.pass_indexes = pass_indexes	1✔
72	super().__init__(localmodel_relation_name + str(NeuObj.count))	1✔
73	self.json['Functions'][self.name] = {}	1✔
74	if input_function is not None:	1✔
75	check(callable(input_function), TypeError, 'The input_function must be callable')	1✔
76	self.input_function = input_function	1✔
77	if output_function is not None:	1✔
78	check(callable(output_function), TypeError, 'The output_function must be callable')	1✔
79	self.output_function = output_function	1✔
80
81	@enforce_types	1✔
82	def __call__(self, inputs:Stream\|tuple, activations:Stream\|tuple= None):	1✔
83	out_sum = []	1✔
84	if type(activations) is not tuple:	1✔
85	activations = (activations,)	1✔
86	self.___activations_matrix(activations,inputs,out_sum)	1✔
87
88	out = out_sum[0]	1✔
89	for ind in range(1,len(out_sum)):	1✔
90	out = out + out_sum[ind]	1✔
91	return out	1✔
92
93	# Definisci una funzione ricorsiva per annidare i cicli for
94	def ___activations_matrix(self, activations, inputs, out, idx=0, idx_list=[]):	1✔
95	if idx != len(activations):	1✔
96	for i in range(activations[idx].dim['dim']):	1✔
97	self.___activations_matrix(activations, inputs, out, idx+1, idx_list+[i])	1✔
98	else:
99	if self.input_function is not None:	1✔
100	if len(inspect.getfullargspec(self.input_function).args) == 0:	1✔
101	if type(inputs) is tuple:	1✔
102	out_in = self.input_function()(*inputs)	×
103	else:
104	out_in = self.input_function()(inputs)	1✔
105	else:
106	if self.pass_indexes:	1✔
107	if type(inputs) is tuple:	1✔
108	out_in = self.input_function(idx_list)(*inputs)	×
109	else:
110	out_in = self.input_function(idx_list)(inputs)	1✔
111	else:
112	if type(inputs) is tuple:	×
113	out_in = self.input_function(*inputs)	×
114	else:
115	out_in = self.input_function(inputs)	×
116	else:
117	check(type(inputs) is not tuple, TypeError, 'The input cannot be a tuple without input_function')	1✔
118	out_in = inputs	1✔
119
120	act = Select(activations[0], idx_list[0])	1✔
121	for ind, i in enumerate(idx_list[1:]):	1✔
122	act = act * Select(activations[ind+1], i)	1✔
123
124	prod = out_in * act	1✔
125
126	if self.output_function is not None:	1✔
127	if len(inspect.signature(self.output_function).parameters) == 0:	1✔
NEW 128	out.append(self.output_function()(prod))	×
129	else:
130	if self.pass_indexes:	1✔
131	out.append(self.output_function(idx_list)(prod))	1✔
132	else:
133	out.append(self.output_function(prod))	1✔
134	else:
135	out.append(prod)	1✔

tonegas / nnodely / 13504608544

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

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