16502811447

Committed 24 Jul 2025 04:44PM UTC coverage: 97.767% (+0.1%) from 97.651%

Build # 16502811447

Build Type

push

github

Committed by

web-flow

Commit Message

New version 1.5.0

This pull request introduces version 1.5.0 of **nnodely**, featuring several updates:
1. Improved clarity of documentation and examples.
2. Support for managing multi-dataset features is now available.
3. DataFrames can now be used to create datasets.
4. Datasets can now be resampled.
5. Random data training has been fixed for both classic and recurrent training.
6. The `state` variable has been removed.
7. It is now possible to add or remove a connection or a closed loop.
8. Partial models can now be exported.
9. The `train` function and the result analysis have been separated.
10. A new function, `trainAndAnalyse`, is now available.
11. The report now works across all network types.
12. The training function code has been reorganized.

Run Details

2901 of 2967 new or added lines in 53 files covered. (97.78%)

16 existing lines in 6 files now uncovered.

12652 of 12941 relevant lines covered (97.77%)

0.98 hits per line

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

93.46

/nnodely/layers/input.py

import copy

from nnodely.basic.relation import NeuObj, Stream, ToStream
from nnodely.support.utils import check, enforce_types
from nnodely.support.jsonutils import merge, stream_to_str
from nnodely.layers.part import SamplePart, TimePart
from nnodely.layers.timeoperation import Derivate, Integrate

class Input(NeuObj):
    """
    Represents an Input in the neural network model.

    .. image:: https://colab.research.google.com/assets/colab-badge.svg
        :target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/states.ipynb
        :alt: Open in Colab

    Parameters
    ----------
    json_name : str
        The name of the JSON field to store the Input configuration.
    name : str
        The name of the Input.
    dimensions : int, optional
        The number of dimensions for the input. Default is 1.

    Attributes
    ----------
    json_name : str
        The name of the JSON field to store the input configuration.
    name : str
        The name of the Input.
    dim : dict
        A dictionary containing the dimensions of the Input.
    json : dict
        A dictionary containing the configuration of the Input.
    """
    def __init__(self, name:str, *, dimensions:int = 1):
        """
        Initializes the Input object.

        Parameters
        ----------
        json_name : str
            The name of the JSON field to store the Input configuration.
        name : str
            The name of the Input.
        dimensions : int, optional
            The number of dimensions for the Input. Default is 1.
        """
        NeuObj.__init__(self, name)
        check(type(dimensions) == int, TypeError,"The dimensions must be a integer")
        self.json['Inputs'][self.name] = {'dim': dimensions }
        self.dim = {'dim': dimensions}

    @enforce_types
    def tw(self, tw:int|float|list, offset:int|float|None = None) -> Stream:
        """
        Selects a time window for the Input.

        Parameters
        ----------
        tw : list or float
            The time window. If a list, it should contain the start and end values. If a float, it represents the time window size.
        offset : float, optional
            The offset for the time window. Default is None.

        Returns
        -------
        Stream
            A Stream representing the TimePart object with the selected time window.

        Raises
        ------
        ValueError
            If the time window is not positive.
        IndexError
            If the offset is not within the time window.
        """
        dim = copy.deepcopy(self.dim)
        json = copy.deepcopy(self.json)
        if type(tw) is list:
            check(len(tw) == 2, TypeError, "The time window must be a list of two elements.")
            check(tw[1] > tw[0], ValueError, "The dimension of the sample window must be positive")
            json['Inputs'][self.name]['tw'] = tw
            tw = tw[1] - tw[0]
        else:
            json['Inputs'][self.name]['tw'] = [-tw, 0]
        check(tw > 0, ValueError, "The time window must be positive")
        dim['tw'] = tw
        if offset is not None:
            check(json['Inputs'][self.name]['tw'][0] <= offset < json['Inputs'][self.name]['tw'][1],
                  IndexError,
                  "The offset must be inside the time window")
        return TimePart(Stream(self.name, json, dim), json['Inputs'][self.name]['tw'][0], json['Inputs'][self.name]['tw'][1], offset)


    @enforce_types
    def sw(self, sw:int|list, offset:int|None = None) -> Stream:
        """
        Selects a sample window for the Input.

        Parameters
        ----------
        sw : list, int
            The sample window. If a list, it should contain the start and end indices. If an int, it represents the number of steps in the past.
        offset : int, optional
            The offset for the sample window. Default is None.

        Returns
        -------
        Stream
            A Stream representing the SamplePart object with the selected samples.

        Raises
        ------
        TypeError
            If the sample window is not an integer or a list of integers.

        Examples
        --------
        Select a sample window considering a signal T = [-3,-2,-1,0,1,2] where the time vector 0 represent the last passed instant. If sw is an integer #1 represent the number of step in the past
            >>> T.s(2) #= [-1, 0] represents two sample step in the past

        If sw is a list [#1,#2] the numbers represent the sample indexes in the vector with the second element excluded
            >>> T.s([-2,0])  #= [-1, 0] represents two time step in the past zero in the future
            >>> T.s([0,1])   #= [1]     the first time in the future
            >>> T.s([-4,-2]) #= [-3,-2]

        The total number of samples can be computed #2-#1. The offset represent the index of the vector that need to be used to offset the window
            >>> T.s(2,offset=-2)       #= [0, 1]      the value of the window is [-1,0]
            >>> T.s([-2,2],offset=-1)  #= [-1,0,1,2]  the value of the window is [-1,0,1,2]
        """
        dim = copy.deepcopy(self.dim)
        json = copy.deepcopy(self.json)
        if type(sw) is list:
            check(len(sw) == 2, TypeError, "The sample window must be a list of two elements.")
            check(type(sw[0]) == int and type(sw[1]) == int, TypeError, "The sample window must be integer")
            check(sw[1] > sw[0], ValueError, "The dimension of the sample window must be positive")
            json['Inputs'][self.name]['sw'] = sw
            sw = sw[1] - sw[0]
        else:
            check(type(sw) == int, TypeError, "The sample window must be integer")
            json['Inputs'][self.name]['sw'] = [-sw, 0]
        check(sw > 0, ValueError, "The sample window must be positive")
        dim['sw'] = sw
        if offset is not None:
            check(json['Inputs'][self.name]['sw'][0] <= offset < json['Inputs'][self.name]['sw'][1],
                  IndexError,
                  "The offset must be inside the sample window")
        return SamplePart(Stream(self.name, json, dim), json['Inputs'][self.name]['sw'][0], json['Inputs'][self.name]['sw'][1], offset)

    @enforce_types
    def z(self, delay:int) -> Stream:
        """
        Considering the Zeta transform notation. The function is used to selects a unitary delay from the Input.

        Parameters
        ----------
        delay : int
            The delay value.

        Returns
        -------
        Stream
            A Stream representing the SamplePart object with the selected delay.

        Examples
        --------
        Select the unitary delay considering a signal T = [-3,-2,-1,0,1,2], where the time vector 0 represent the last passed instant
            T.z(-1) = 1
            T.z(0)  = 0 # the last passed instant
            T.z(2)  = -2
        """
        dim = copy.deepcopy(self.dim)
        json = copy.deepcopy(self.json)
        sw = [(-delay) - 1, (-delay)]
        json['Inputs'][self.name]['sw'] = sw
        dim['sw'] = sw[1] - sw[0]
        return SamplePart(Stream(self.name, json, dim), json['Inputs'][self.name]['sw'][0], json['Inputs'][self.name]['sw'][1], None)

    @enforce_types
    def last(self) -> Stream:
        """
        Selects the last passed instant for the input.

        Returns
        -------
        Stream
            A Stream representing the SamplePart object with the last passed instant.
        """
        return self.z(0)

    @enforce_types
    def next(self) -> Stream:
        """
        Selects the next instant for the input.

        Returns
        -------
        Stream
            A Stream representing the SamplePart object with the next instant.
        """
        return self.z(-1)

    @enforce_types
    def s(self, order:int, *, der_name:str|None = None, int_name:str|None = None, method:str = 'euler') -> Stream:
        """
        Considering the Laplace transform notation. The function is used to operate an integral or derivate operation on the input.
        The order of the integral or the derivative operation is indicated by the order parameter.

        Parameters
        ----------
        order : int
            Order of the Laplace transform
        method : str, optional
            Integration or derivation method

        Returns
        -------
        Stream
            A Stream of the signal represents the integral or derivation operation.
        """
        check(order != 0, ValueError, "The order must be a positive or negative integer not a zero")
        if order > 0:
            o = self.last()
            for i in range(order):
                o = Derivate(o, der_name = der_name, int_name = int_name, method = method)
        elif order < 0:
            o = self.last()
            for i in range(-order):
                o = Integrate(o, der_name = der_name, int_name = int_name, method = method)
        return o

    @enforce_types
    def connect(self, obj:Stream) -> "Input":
        """
        Update and return the current Input with a given Stream object.

        Parameters
        ----------
        obj : Stream
            The Stream object for update the Input.

        Returns
        -------
        Input
            A Input with the connection to the obj Stream

        Raises
        ------
        TypeError
            If the provided object is not of type Input.
        KeyError
            If the Input variable is already connected.
        """
        check(type(obj) is Stream, TypeError,
              f"The {obj} must be a Stream and not a {type(obj)}.")
        self.json = merge(self.json, obj.json)
        check('closedLoop' not in self.json['Inputs'][self.name] or 'connect' not in self.json['Inputs'][self.name], KeyError,
              f"The Input variable {self.name} is already connected.")
        self.json['Inputs'][self.name]['connect'] = obj.name
        self.json['Inputs'][self.name]['local'] = 1
        return self

    @enforce_types
    def closedLoop(self, obj:Stream) -> "Input":
        """
        Update and return the current Input in a closed loop with a given Stream object.

        Parameters
        ----------
        obj : Stream
            The Stream object for update the Input.

        Returns
        -------
        Input
            A Input with the connection to the obj Stream

        Raises
        ------
        TypeError
            If the provided object is not of type Input.
        KeyError
            If the Input variable is already connected.
        """
        from nnodely.layers.input import Input
        check(type(obj) is Stream, TypeError,
              f"The {obj} must be a Stream and not a {type(obj)}.")
        self.json = merge(self.json, obj.json)
        check('closedLoop' not in self.json['Inputs'][self.name] or 'connect' not in self.json['Inputs'][self.name],
              KeyError,
              f"The Input variable {self.name} is already connected.")
        self.json['Inputs'][self.name]['closedLoop'] = self.name
        self.json['Inputs'][self.name]['local'] = 1
        return self

    def __str__(self):
        return stream_to_str(self, 'Input')

    def __repr__(self):
        return self.__str__()

# connect operation
connect_name = 'connect'
closedloop_name = 'closedLoop'

class Connect(Stream, ToStream):
    @enforce_types
    def __init__(self, obj1:Stream, obj2:Input, *, local:bool=False) -> Stream:
        super().__init__(obj1.name,merge(obj1.json, obj2.json),obj1.dim)
        check(closedloop_name not in self.json['Inputs'][obj2.name] or connect_name not in self.json['Inputs'][obj2.name],
              KeyError,f"The input variable {obj2.name} is already connected.")
        self.json['Inputs'][obj2.name][connect_name] = obj1.name
        self.json['Inputs'][obj2.name]['local'] = int(local)

class ClosedLoop(Stream, ToStream):
    @enforce_types
    def __init__(self, obj1:Stream, obj2:Input, *, local:bool=False) -> Stream:
        super().__init__(obj1.name, merge(obj1.json, obj2.json), obj1.dim)
        check(closedloop_name not in self.json['Inputs'][obj2.name] or connect_name not in self.json['Inputs'][obj2.name],
              KeyError, f"The input variable {obj2.name} is already connected.")
        self.json['Inputs'][obj2.name][closedloop_name] = obj1.name
        self.json['Inputs'][obj2.name]['local'] = int(local)

1	import copy	1✔
2
3	from nnodely.basic.relation import NeuObj, Stream, ToStream	1✔
4	from nnodely.support.utils import check, enforce_types	1✔
5	from nnodely.support.jsonutils import merge, stream_to_str	1✔
6	from nnodely.layers.part import SamplePart, TimePart	1✔
7	from nnodely.layers.timeoperation import Derivate, Integrate	1✔
8
9	class Input(NeuObj):	1✔
10	"""
11	Represents an Input in the neural network model.
12
13	.. image:: https://colab.research.google.com/assets/colab-badge.svg
14	:target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/states.ipynb
15	:alt: Open in Colab
16
17	Parameters
18	----------
19	json_name : str
20	The name of the JSON field to store the Input configuration.
21	name : str
22	The name of the Input.
23	dimensions : int, optional
24	The number of dimensions for the input. Default is 1.
25
26	Attributes
27	----------
28	json_name : str
29	The name of the JSON field to store the input configuration.
30	name : str
31	The name of the Input.
32	dim : dict
33	A dictionary containing the dimensions of the Input.
34	json : dict
35	A dictionary containing the configuration of the Input.
36	"""
37	def __init__(self, name:str, *, dimensions:int = 1):	1✔
38	"""
39	Initializes the Input object.
40
41	Parameters
42	----------
43	json_name : str
44	The name of the JSON field to store the Input configuration.
45	name : str
46	The name of the Input.
47	dimensions : int, optional
48	The number of dimensions for the Input. Default is 1.
49	"""
50	NeuObj.__init__(self, name)	1✔
51	check(type(dimensions) == int, TypeError,"The dimensions must be a integer")	1✔
52	self.json['Inputs'][self.name] = {'dim': dimensions }	1✔
53	self.dim = {'dim': dimensions}	1✔
54
55	@enforce_types	1✔
56	def tw(self, tw:int\|float\|list, offset:int\|float\|None = None) -> Stream:	1✔
57	"""
58	Selects a time window for the Input.
59
60	Parameters
61	----------
62	tw : list or float
63	The time window. If a list, it should contain the start and end values. If a float, it represents the time window size.
64	offset : float, optional
65	The offset for the time window. Default is None.
66
67	Returns
68	-------
69	Stream
70	A Stream representing the TimePart object with the selected time window.
71
72	Raises
73	------
74	ValueError
75	If the time window is not positive.
76	IndexError
77	If the offset is not within the time window.
78	"""
79	dim = copy.deepcopy(self.dim)	1✔
80	json = copy.deepcopy(self.json)	1✔
81	if type(tw) is list:	1✔
82	check(len(tw) == 2, TypeError, "The time window must be a list of two elements.")	1✔
83	check(tw[1] > tw[0], ValueError, "The dimension of the sample window must be positive")	1✔
84	json['Inputs'][self.name]['tw'] = tw	1✔
85	tw = tw[1] - tw[0]	1✔
86	else:
87	json['Inputs'][self.name]['tw'] = [-tw, 0]	1✔
88	check(tw > 0, ValueError, "The time window must be positive")	1✔
89	dim['tw'] = tw	1✔
90	if offset is not None:	1✔
91	check(json['Inputs'][self.name]['tw'][0] <= offset < json['Inputs'][self.name]['tw'][1],	1✔
92	IndexError,
93	"The offset must be inside the time window")
94	return TimePart(Stream(self.name, json, dim), json['Inputs'][self.name]['tw'][0], json['Inputs'][self.name]['tw'][1], offset)	1✔
95
96
97	@enforce_types	1✔
98	def sw(self, sw:int\|list, offset:int\|None = None) -> Stream:	1✔
99	"""
100	Selects a sample window for the Input.
101
102	Parameters
103	----------
104	sw : list, int
105	The sample window. If a list, it should contain the start and end indices. If an int, it represents the number of steps in the past.
106	offset : int, optional
107	The offset for the sample window. Default is None.
108
109	Returns
110	-------
111	Stream
112	A Stream representing the SamplePart object with the selected samples.
113
114	Raises
115	------
116	TypeError
117	If the sample window is not an integer or a list of integers.
118
119	Examples
120	--------
121	Select a sample window considering a signal T = [-3,-2,-1,0,1,2] where the time vector 0 represent the last passed instant. If sw is an integer #1 represent the number of step in the past
122	>>> T.s(2) #= [-1, 0] represents two sample step in the past
123
124	If sw is a list [#1,#2] the numbers represent the sample indexes in the vector with the second element excluded
125	>>> T.s([-2,0]) #= [-1, 0] represents two time step in the past zero in the future
126	>>> T.s([0,1]) #= [1] the first time in the future
127	>>> T.s([-4,-2]) #= [-3,-2]
128
129	The total number of samples can be computed #2-#1. The offset represent the index of the vector that need to be used to offset the window
130	>>> T.s(2,offset=-2) #= [0, 1] the value of the window is [-1,0]
131	>>> T.s([-2,2],offset=-1) #= [-1,0,1,2] the value of the window is [-1,0,1,2]
132	"""
133	dim = copy.deepcopy(self.dim)	1✔
134	json = copy.deepcopy(self.json)	1✔
135	if type(sw) is list:	1✔
136	check(len(sw) == 2, TypeError, "The sample window must be a list of two elements.")	1✔
137	check(type(sw[0]) == int and type(sw[1]) == int, TypeError, "The sample window must be integer")	1✔
138	check(sw[1] > sw[0], ValueError, "The dimension of the sample window must be positive")	1✔
139	json['Inputs'][self.name]['sw'] = sw	1✔
140	sw = sw[1] - sw[0]	1✔
141	else:
142	check(type(sw) == int, TypeError, "The sample window must be integer")	1✔
143	json['Inputs'][self.name]['sw'] = [-sw, 0]	1✔
144	check(sw > 0, ValueError, "The sample window must be positive")	1✔
145	dim['sw'] = sw	1✔
146	if offset is not None:	1✔
147	check(json['Inputs'][self.name]['sw'][0] <= offset < json['Inputs'][self.name]['sw'][1],	1✔
148	IndexError,
149	"The offset must be inside the sample window")
150	return SamplePart(Stream(self.name, json, dim), json['Inputs'][self.name]['sw'][0], json['Inputs'][self.name]['sw'][1], offset)	1✔
151
152	@enforce_types	1✔
153	def z(self, delay:int) -> Stream:	1✔
154	"""
155	Considering the Zeta transform notation. The function is used to selects a unitary delay from the Input.
156
157	Parameters
158	----------
159	delay : int
160	The delay value.
161
162	Returns
163	-------
164	Stream
165	A Stream representing the SamplePart object with the selected delay.
166
167	Examples
168	--------
169	Select the unitary delay considering a signal T = [-3,-2,-1,0,1,2], where the time vector 0 represent the last passed instant
170	T.z(-1) = 1
171	T.z(0) = 0 # the last passed instant
172	T.z(2) = -2
173	"""
174	dim = copy.deepcopy(self.dim)	1✔
175	json = copy.deepcopy(self.json)	1✔
176	sw = [(-delay) - 1, (-delay)]	1✔
177	json['Inputs'][self.name]['sw'] = sw	1✔
178	dim['sw'] = sw[1] - sw[0]	1✔
179	return SamplePart(Stream(self.name, json, dim), json['Inputs'][self.name]['sw'][0], json['Inputs'][self.name]['sw'][1], None)	1✔
180
181	@enforce_types	1✔
182	def last(self) -> Stream:	1✔
183	"""
184	Selects the last passed instant for the input.
185
186	Returns
187	-------
188	Stream
189	A Stream representing the SamplePart object with the last passed instant.
190	"""
191	return self.z(0)	1✔
192
193	@enforce_types	1✔
194	def next(self) -> Stream:	1✔
195	"""
196	Selects the next instant for the input.
197
198	Returns
199	-------
200	Stream
201	A Stream representing the SamplePart object with the next instant.
202	"""
203	return self.z(-1)	1✔
204
205	@enforce_types	1✔
206	def s(self, order:int, *, der_name:str\|None = None, int_name:str\|None = None, method:str = 'euler') -> Stream:	1✔
207	"""
208	Considering the Laplace transform notation. The function is used to operate an integral or derivate operation on the input.
209	The order of the integral or the derivative operation is indicated by the order parameter.
210
211	Parameters
212	----------
213	order : int
214	Order of the Laplace transform
215	method : str, optional
216	Integration or derivation method
217
218	Returns
219	-------
220	Stream
221	A Stream of the signal represents the integral or derivation operation.
222	"""
223	check(order != 0, ValueError, "The order must be a positive or negative integer not a zero")	1✔
224	if order > 0:	1✔
225	o = self.last()	1✔
226	for i in range(order):	1✔
227	o = Derivate(o, der_name = der_name, int_name = int_name, method = method)	1✔
228	elif order < 0:	1✔
229	o = self.last()	1✔
230	for i in range(-order):	1✔
231	o = Integrate(o, der_name = der_name, int_name = int_name, method = method)	1✔
232	return o	1✔
233
234	@enforce_types	1✔
235	def connect(self, obj:Stream) -> "Input":	1✔
236	"""
237	Update and return the current Input with a given Stream object.
238
239	Parameters
240	----------
241	obj : Stream
242	The Stream object for update the Input.
243
244	Returns
245	-------
246	Input
247	A Input with the connection to the obj Stream
248
249	Raises
250	------
251	TypeError
252	If the provided object is not of type Input.
253	KeyError
254	If the Input variable is already connected.
255	"""
256	check(type(obj) is Stream, TypeError,	1✔
257	f"The {obj} must be a Stream and not a {type(obj)}.")
258	self.json = merge(self.json, obj.json)	1✔
259	check('closedLoop' not in self.json['Inputs'][self.name] or 'connect' not in self.json['Inputs'][self.name], KeyError,	1✔
260	f"The Input variable {self.name} is already connected.")
261	self.json['Inputs'][self.name]['connect'] = obj.name	1✔
262	self.json['Inputs'][self.name]['local'] = 1	1✔
263	return self	1✔
264
265	@enforce_types	1✔
266	def closedLoop(self, obj:Stream) -> "Input":	1✔
267	"""
268	Update and return the current Input in a closed loop with a given Stream object.
269
270	Parameters
271	----------
272	obj : Stream
273	The Stream object for update the Input.
274
275	Returns
276	-------
277	Input
278	A Input with the connection to the obj Stream
279
280	Raises
281	------
282	TypeError
283	If the provided object is not of type Input.
284	KeyError
285	If the Input variable is already connected.
286	"""
NEW 287	from nnodely.layers.input import Input	×
NEW 288	check(type(obj) is Stream, TypeError,	×
289	f"The {obj} must be a Stream and not a {type(obj)}.")
NEW 290	self.json = merge(self.json, obj.json)	×
NEW 291	check('closedLoop' not in self.json['Inputs'][self.name] or 'connect' not in self.json['Inputs'][self.name],	×
292	KeyError,
293	f"The Input variable {self.name} is already connected.")
NEW 294	self.json['Inputs'][self.name]['closedLoop'] = self.name	×
NEW 295	self.json['Inputs'][self.name]['local'] = 1	×
NEW 296	return self	×
297
298	def __str__(self):	1✔
299	return stream_to_str(self, 'Input')	1✔
300
301	def __repr__(self):	1✔
302	return self.__str__()	1✔
303
304	# connect operation
305	connect_name = 'connect'	1✔
306	closedloop_name = 'closedLoop'	1✔
307
308	class Connect(Stream, ToStream):	1✔
309	@enforce_types	1✔
310	def __init__(self, obj1:Stream, obj2:Input, *, local:bool=False) -> Stream:	1✔
311	super().__init__(obj1.name,merge(obj1.json, obj2.json),obj1.dim)	1✔
312	check(closedloop_name not in self.json['Inputs'][obj2.name] or connect_name not in self.json['Inputs'][obj2.name],	1✔
313	KeyError,f"The input variable {obj2.name} is already connected.")
314	self.json['Inputs'][obj2.name][connect_name] = obj1.name	1✔
315	self.json['Inputs'][obj2.name]['local'] = int(local)	1✔
316
317	class ClosedLoop(Stream, ToStream):	1✔
318	@enforce_types	1✔
319	def __init__(self, obj1:Stream, obj2:Input, *, local:bool=False) -> Stream:	1✔
320	super().__init__(obj1.name, merge(obj1.json, obj2.json), obj1.dim)	1✔
321	check(closedloop_name not in self.json['Inputs'][obj2.name] or connect_name not in self.json['Inputs'][obj2.name],	1✔
322	KeyError, f"The input variable {obj2.name} is already connected.")
323	self.json['Inputs'][obj2.name][closedloop_name] = obj1.name	1✔
324	self.json['Inputs'][obj2.name]['local'] = int(local)	1✔

tonegas / nnodely / 16502811447

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