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

94.55

/nnodely/basic/modeldef.py

import copy

import numpy as np

from nnodely.support.utils import check, check_and_get_list
from nnodely.support.jsonutils import merge, subjson_from_model,subjson_from_relation, check_model, get_models_json
from nnodely.basic.relation import MAIN_JSON, Stream, check_names
from nnodely.layers.output import Output

from nnodely.support.logger import logging, nnLogger
log = nnLogger(__name__, logging.INFO)


class ModelDef:
    def __init__(self, model_def = MAIN_JSON):
        # Models definition
        self.__json_base = copy.deepcopy(model_def)

        # Initialize the model definition
        self.__json = copy.deepcopy(self.__json_base)
        if "SampleTime" in self.__json['Info']:
            self.__sample_time = self.__json['Info']["SampleTime"]
        else:
            self.__sample_time = None

    def __contains__(self, key):
        return key in self.__json

    def __getitem__(self, key):
        if key in self.__json:
            return self.__json[key]
        else:
            return None

    def __setitem__(self, key, value):
        self.__json[key] = value

    def __rebuild_json(self, models_names, minimizers):
        new_json = subjson_from_model(self.__json, list(models_names))

        if 'Minimizers' in self.__json:
            rel_A = [self.__json['Minimizers'][key]['A'] for key in minimizers]
            rel_B = [self.__json['Minimizers'][key]['B'] for key in minimizers]
            relations_name = set(rel_A) | set(rel_B)
            if len(relations_name) != 0:
                minimizers_json = subjson_from_relation(self.__json, list(relations_name))
                new_json = merge(new_json, minimizers_json)

        return copy.deepcopy(new_json)

    def recurrentInputs(self):
        return {key:value for key, value in self.__json['Inputs'].items() if ('closedLoop' in value.keys() or 'connect' in value.keys())}

    def getJson(self, models:list|str|None = None) -> dict:
        if models is None:
            return copy.deepcopy(self.__json)
        else:
            json = subjson_from_model(self.__json, models)
            check_model(json)
            return copy.deepcopy(json)

    def getSampleTime(self):
        check(self.__sample_time is not None, AttributeError, "Sample time is not defined the model is not neuralized!")
        return self.__sample_time

    def isDefined(self):
        return self.__json is not None

    def addConnect(self, stream_name:str, input_name:str, local:bool = False):
        input_name = check_and_get_list(input_name, set(self.__json['Inputs'].keys()),
                                       lambda name: f"The name {name} is not part of the available inputs")[0]
        stream_name = check_and_get_list(stream_name, set(self.__json['Relations'].keys()),
                                        lambda name: f"The name {name} is not part of the available relations")[0]
        self.__json['Inputs'][input_name]['connect'] = stream_name
        self.__json['Inputs'][input_name]['local'] = int(local)

    def addClosedLoop(self, stream_name:str, input_name:str, local:bool = False):
        input_name = check_and_get_list(input_name, set(self.__json['Inputs'].keys()),
                                       lambda name: f"The name {name} is not part of the available inputs")[0]
        stream_name = check_and_get_list(stream_name, set(self.__json['Relations'].keys()),
                                        lambda name: f"The name {name} is not part of the available relations")[0]
        self.__json['Inputs'][input_name]['closedLoop'] = stream_name
        self.__json['Inputs'][input_name]['local'] = int(local)

    def removeConnection(self, name_list:str|list[str]):
        name_list = check_and_get_list(name_list, set(self.__json['Inputs'].keys()), lambda name: f"The name {name} is not part of the available inputs")
        for input_in in name_list:
            if 'closedLoop' in self.__json['Inputs'][input_in].keys():
                del self.__json['Inputs'][input_in]['closedLoop']
                del self.__json['Inputs'][input_in]['local']
            elif 'connect' in self.__json['Inputs'][input_in].keys():
                del self.__json['Inputs'][input_in]['connect']
                del self.__json['Inputs'][input_in]['local']
            else:
                raise ValueError(f"The input '{input_in}' has no connection or closed loop defined")

    def addModel(self, name:str, stream_list):
        if isinstance(stream_list, Output):
            stream_list = [stream_list]

        json = MAIN_JSON
        for stream in stream_list:
            json = merge(json, stream.json)
        check_model(json)

        if 'Models' not in self.__json:
            self.__json = merge(self.__json, json)
            self.__json['Models'] = name
        else:
            models_names = set((self.__json['Models'],)) if type(self.__json['Models']) is str else set(self.__json['Models'].keys())
            check_names(name, models_names, 'Models')
            if type(self.__json['Models']) is str:
                self.__json['Models'] = {self.__json['Models']: get_models_json(self.__json)}
            self.__json = merge(self.__json, json)
            self.__json['Models'][name] = get_models_json(json)

    def removeModel(self, name_list):
        if 'Models' not in self.__json:
            raise ValueError("No Models are defined")
        models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())
        name_list = check_and_get_list(name_list, models_names, lambda name: f"The name {name} is not part of the available models")
        models_names -= set(name_list)
        minimizers = set(self.__json['Minimizers'].keys()) if 'Minimizers' in self.__json else None
        self.__json = self.__rebuild_json(models_names, minimizers)

    def addMinimize(self, name, streamA, streamB, loss_function='mse'):
        check(isinstance(streamA, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')
        check(isinstance(streamB, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')
        # check(streamA.dim == streamB.dim, ValueError, f'Dimension of streamA={streamA.dim} and streamB={streamB.dim} are not equal.')
        if 'Minimizers' not in self.__json:
            self.__json['Minimizers'] = {}

        check_names(name, set(self.__json['Minimizers'].keys()), 'Minimizers')
        streams = merge(streamA.json, streamB.json)
        streamA_name = streamA.json['Outputs'][streamA.name] if isinstance(streamA, Output) else streamA.name
        streamB_name = streamB.json['Outputs'][streamB.name] if isinstance(streamB, Output) else streamB.name
        self.__json = merge(self.__json, streams)
        self.__json['Minimizers'][name] = {}
        self.__json['Minimizers'][name]['A'] = streamA_name
        self.__json['Minimizers'][name]['B'] = streamB_name
        self.__json['Minimizers'][name]['loss'] = loss_function

    def removeMinimize(self, name_list):
        if 'Minimizers' not in self.__json:
            raise ValueError("No Minimizers are defined")
        name_list = check_and_get_list(name_list, self.__json['Minimizers'].keys(), lambda name: f"The name {name} is not part of the available minimizers")
        models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())
        remaining_minimizers = set(self.__json['Minimizers'].keys()) - set(name_list) if 'Minimizers' in self.__json else None
        self.__json = self.__rebuild_json(models_names, remaining_minimizers)

    def setBuildWindow(self, sample_time = None):
        check(self.__json is not None, RuntimeError, "No model is defined!")
        if sample_time is not None:
            check(sample_time > 0, RuntimeError, 'Sample time must be strictly positive!')
            self.__sample_time = sample_time
        else:
            if self.__sample_time is None:
                self.__sample_time = 1

        self.__json['Info'] = {"SampleTime": self.__sample_time}

        check(self.__json['Inputs'] != {}, RuntimeError, "No model is defined!")
        json_inputs = self.__json['Inputs']

        input_ns_backward, input_ns_forward = {}, {}
        for key, value in json_inputs.items():
            if 'sw' not in value and 'tw' not in value:
                assert False, f"Input '{key}' has no time window or sample window"
            if 'sw' not in value and self.__sample_time is not None:
                ## check if value['tw'] is a multiple of sample_time
                absolute_tw = abs(value['tw'][0]) + abs(value['tw'][1])
                check(round(absolute_tw % self.__sample_time) == 0, ValueError,
                      f"Time window of input '{key}' is not a multiple of sample time. This network cannot be neuralized")
                input_ns_backward[key] = round(-value['tw'][0] / self.__sample_time)
                input_ns_forward[key] = round(value['tw'][1] / self.__sample_time)
            elif self.__sample_time is not None:
                if 'tw' in value:
                    input_ns_backward[key] = max(round(-value['tw'][0] / self.__sample_time), -value['sw'][0])
                    input_ns_forward[key] = max(round(value['tw'][1] / self.__sample_time), value['sw'][1])
                else:
                    input_ns_backward[key] = -value['sw'][0]
                    input_ns_forward[key] =  value['sw'][1]
            else:
                check(value['tw'] == [0,0], RuntimeError, f"Sample time is not defined for input '{key}'")
                input_ns_backward[key] = -value['sw'][0]
                input_ns_forward[key] = value['sw'][1]
            value['ns'] = [input_ns_backward[key], input_ns_forward[key]]
            value['ntot'] = sum(value['ns'])

        self.__json['Info']['ns'] = [max(input_ns_backward.values()), max(input_ns_forward.values())]
        self.__json['Info']['ntot'] = sum(self.__json['Info']['ns'])
        if self.__json['Info']['ns'][0] < 0:
            log.warning(
                f"The input is only in the far past the max_samples_backward is: {self.__json['Info']['ns'][0]}")
        if self.__json['Info']['ns'][1] < 0:
            log.warning(
                f"The input is only in the far future the max_sample_forward is: {self.__json['Info']['ns'][1]}")

        for k, v in (self.__json['Parameters'] | self.__json['Constants']).items():
            if 'values' in v:
                window = 'tw' if 'tw' in v.keys() else ('sw' if 'sw' in v.keys() else None)
                if window == 'tw':
                    check(np.array(v['values']).shape[0] == v['tw'] / self.__sample_time, ValueError,
                      f"{k} has a different number of values for this sample time.")
                if v['values'] == "SampleTime":
                    v['values'] = self.__sample_time

    def updateParameters(self, model = None, *, clear_model = False):
        if clear_model:
            for key in self.__json['Parameters'].keys():
                if 'init_values' in self.__json['Parameters'][key]:
                    self.__json['Parameters'][key]['values'] = self.__json['Parameters'][key]['init_values']
                elif 'values' in self.__json['Parameters'][key]:
                    del self.__json['Parameters'][key]['values']
        elif model is not None:
            for key in self.__json['Parameters'].keys():
                if key in model.all_parameters:
                    self.__json['Parameters'][key]['values'] = model.all_parameters[key].tolist()


1	import copy	1✔
2
3	import numpy as np	1✔
4
5	from nnodely.support.utils import check, check_and_get_list	1✔
6	from nnodely.support.jsonutils import merge, subjson_from_model,subjson_from_relation, check_model, get_models_json	1✔
7	from nnodely.basic.relation import MAIN_JSON, Stream, check_names	1✔
8	from nnodely.layers.output import Output	1✔
9
10	from nnodely.support.logger import logging, nnLogger	1✔
11	log = nnLogger(__name__, logging.INFO)	1✔
12
13
14	class ModelDef:	1✔
15	def __init__(self, model_def = MAIN_JSON):	1✔
16	# Models definition
17	self.__json_base = copy.deepcopy(model_def)	1✔
18
19	# Initialize the model definition
20	self.__json = copy.deepcopy(self.__json_base)	1✔
21	if "SampleTime" in self.__json['Info']:	1✔
22	self.__sample_time = self.__json['Info']["SampleTime"]	1✔
23	else:
24	self.__sample_time = None	1✔
25
26	def __contains__(self, key):	1✔
27	return key in self.__json	1✔
28
29	def __getitem__(self, key):	1✔
30	if key in self.__json:	1✔
31	return self.__json[key]	1✔
32	else:
33	return None	1✔
34
35	def __setitem__(self, key, value):	1✔
36	self.__json[key] = value	×
37
38	def __rebuild_json(self, models_names, minimizers):	1✔
39	new_json = subjson_from_model(self.__json, list(models_names))	1✔
40
41	if 'Minimizers' in self.__json:	1✔
42	rel_A = [self.__json['Minimizers'][key]['A'] for key in minimizers]	1✔
43	rel_B = [self.__json['Minimizers'][key]['B'] for key in minimizers]	1✔
44	relations_name = set(rel_A) \| set(rel_B)	1✔
45	if len(relations_name) != 0:	1✔
NEW 46	minimizers_json = subjson_from_relation(self.__json, list(relations_name))	×
NEW 47	new_json = merge(new_json, minimizers_json)	×
48
49	return copy.deepcopy(new_json)	1✔
50
51	def recurrentInputs(self):	1✔
52	return {key:value for key, value in self.__json['Inputs'].items() if ('closedLoop' in value.keys() or 'connect' in value.keys())}	1✔
53
54	def getJson(self, models:list\|str\|None = None) -> dict:	1✔
55	if models is None:	1✔
56	return copy.deepcopy(self.__json)	1✔
57	else:
58	json = subjson_from_model(self.__json, models)	1✔
59	check_model(json)	1✔
60	return copy.deepcopy(json)	1✔
61
62	def getSampleTime(self):	1✔
63	check(self.__sample_time is not None, AttributeError, "Sample time is not defined the model is not neuralized!")	1✔
64	return self.__sample_time	1✔
65
66	def isDefined(self):	1✔
67	return self.__json is not None	1✔
68
69	def addConnect(self, stream_name:str, input_name:str, local:bool = False):	1✔
70	input_name = check_and_get_list(input_name, set(self.__json['Inputs'].keys()),	1✔
71	lambda name: f"The name {name} is not part of the available inputs")[0]
72	stream_name = check_and_get_list(stream_name, set(self.__json['Relations'].keys()),	1✔
73	lambda name: f"The name {name} is not part of the available relations")[0]
74	self.__json['Inputs'][input_name]['connect'] = stream_name	1✔
75	self.__json['Inputs'][input_name]['local'] = int(local)	1✔
76
77	def addClosedLoop(self, stream_name:str, input_name:str, local:bool = False):	1✔
78	input_name = check_and_get_list(input_name, set(self.__json['Inputs'].keys()),	1✔
79	lambda name: f"The name {name} is not part of the available inputs")[0]
80	stream_name = check_and_get_list(stream_name, set(self.__json['Relations'].keys()),	1✔
81	lambda name: f"The name {name} is not part of the available relations")[0]
82	self.__json['Inputs'][input_name]['closedLoop'] = stream_name	1✔
83	self.__json['Inputs'][input_name]['local'] = int(local)	1✔
84
85	def removeConnection(self, name_list:str\|list[str]):	1✔
86	name_list = check_and_get_list(name_list, set(self.__json['Inputs'].keys()), lambda name: f"The name {name} is not part of the available inputs")	1✔
87	for input_in in name_list:	1✔
88	if 'closedLoop' in self.__json['Inputs'][input_in].keys():	1✔
89	del self.__json['Inputs'][input_in]['closedLoop']	1✔
90	del self.__json['Inputs'][input_in]['local']	1✔
91	elif 'connect' in self.__json['Inputs'][input_in].keys():	1✔
92	del self.__json['Inputs'][input_in]['connect']	1✔
93	del self.__json['Inputs'][input_in]['local']	1✔
94	else:
95	raise ValueError(f"The input '{input_in}' has no connection or closed loop defined")	1✔
96
97	def addModel(self, name:str, stream_list):	1✔
98	if isinstance(stream_list, Output):	1✔
99	stream_list = [stream_list]	1✔
100
101	json = MAIN_JSON	1✔
102	for stream in stream_list:	1✔
103	json = merge(json, stream.json)	1✔
104	check_model(json)	1✔
105
106	if 'Models' not in self.__json:	1✔
107	self.__json = merge(self.__json, json)	1✔
108	self.__json['Models'] = name	1✔
109	else:
110	models_names = set((self.__json['Models'],)) if type(self.__json['Models']) is str else set(self.__json['Models'].keys())	1✔
111	check_names(name, models_names, 'Models')	1✔
112	if type(self.__json['Models']) is str:	1✔
113	self.__json['Models'] = {self.__json['Models']: get_models_json(self.__json)}	1✔
114	self.__json = merge(self.__json, json)	1✔
115	self.__json['Models'][name] = get_models_json(json)	1✔
116
117	def removeModel(self, name_list):	1✔
118	if 'Models' not in self.__json:	1✔
NEW 119	raise ValueError("No Models are defined")	×
120	models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())	1✔
121	name_list = check_and_get_list(name_list, models_names, lambda name: f"The name {name} is not part of the available models")	1✔
122	models_names -= set(name_list)	1✔
123	minimizers = set(self.__json['Minimizers'].keys()) if 'Minimizers' in self.__json else None	1✔
124	self.__json = self.__rebuild_json(models_names, minimizers)	1✔
125
126	def addMinimize(self, name, streamA, streamB, loss_function='mse'):	1✔
127	check(isinstance(streamA, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')	1✔
128	check(isinstance(streamB, (Output, Stream)), TypeError, 'streamA must be an instance of Output or Stream')	1✔
129	# check(streamA.dim == streamB.dim, ValueError, f'Dimension of streamA={streamA.dim} and streamB={streamB.dim} are not equal.')
130	if 'Minimizers' not in self.__json:	1✔
131	self.__json['Minimizers'] = {}	1✔
132
133	check_names(name, set(self.__json['Minimizers'].keys()), 'Minimizers')	1✔
134	streams = merge(streamA.json, streamB.json)	1✔
135	streamA_name = streamA.json['Outputs'][streamA.name] if isinstance(streamA, Output) else streamA.name	1✔
136	streamB_name = streamB.json['Outputs'][streamB.name] if isinstance(streamB, Output) else streamB.name	1✔
137	self.__json = merge(self.__json, streams)	1✔
138	self.__json['Minimizers'][name] = {}	1✔
139	self.__json['Minimizers'][name]['A'] = streamA_name	1✔
140	self.__json['Minimizers'][name]['B'] = streamB_name	1✔
141	self.__json['Minimizers'][name]['loss'] = loss_function	1✔
142
143	def removeMinimize(self, name_list):	1✔
144	if 'Minimizers' not in self.__json:	1✔
NEW 145	raise ValueError("No Minimizers are defined")	×
146	name_list = check_and_get_list(name_list, self.__json['Minimizers'].keys(), lambda name: f"The name {name} is not part of the available minimizers")	1✔
147	models_names = {self.__json['Models']} if type(self.__json['Models']) is str else set(self.__json['Models'].keys())	1✔
148	remaining_minimizers = set(self.__json['Minimizers'].keys()) - set(name_list) if 'Minimizers' in self.__json else None	1✔
149	self.__json = self.__rebuild_json(models_names, remaining_minimizers)	1✔
150
151	def setBuildWindow(self, sample_time = None):	1✔
152	check(self.__json is not None, RuntimeError, "No model is defined!")	1✔
153	if sample_time is not None:	1✔
154	check(sample_time > 0, RuntimeError, 'Sample time must be strictly positive!')	1✔
155	self.__sample_time = sample_time	1✔
156	else:
157	if self.__sample_time is None:	1✔
158	self.__sample_time = 1	1✔
159
160	self.__json['Info'] = {"SampleTime": self.__sample_time}	1✔
161
162	check(self.__json['Inputs'] != {}, RuntimeError, "No model is defined!")	1✔
163	json_inputs = self.__json['Inputs']	1✔
164
165	input_ns_backward, input_ns_forward = {}, {}	1✔
166	for key, value in json_inputs.items():	1✔
167	if 'sw' not in value and 'tw' not in value:	1✔
NEW 168	assert False, f"Input '{key}' has no time window or sample window"	×
169	if 'sw' not in value and self.__sample_time is not None:	1✔
170	## check if value['tw'] is a multiple of sample_time
171	absolute_tw = abs(value['tw'][0]) + abs(value['tw'][1])	1✔
172	check(round(absolute_tw % self.__sample_time) == 0, ValueError,	1✔
173	f"Time window of input '{key}' is not a multiple of sample time. This network cannot be neuralized")
174	input_ns_backward[key] = round(-value['tw'][0] / self.__sample_time)	1✔
175	input_ns_forward[key] = round(value['tw'][1] / self.__sample_time)	1✔
176	elif self.__sample_time is not None:	1✔
177	if 'tw' in value:	1✔
178	input_ns_backward[key] = max(round(-value['tw'][0] / self.__sample_time), -value['sw'][0])	1✔
179	input_ns_forward[key] = max(round(value['tw'][1] / self.__sample_time), value['sw'][1])	1✔
180	else:
181	input_ns_backward[key] = -value['sw'][0]	1✔
182	input_ns_forward[key] = value['sw'][1]	1✔
183	else:
184	check(value['tw'] == [0,0], RuntimeError, f"Sample time is not defined for input '{key}'")	×
185	input_ns_backward[key] = -value['sw'][0]	×
186	input_ns_forward[key] = value['sw'][1]	×
187	value['ns'] = [input_ns_backward[key], input_ns_forward[key]]	1✔
188	value['ntot'] = sum(value['ns'])	1✔
189
190	self.__json['Info']['ns'] = [max(input_ns_backward.values()), max(input_ns_forward.values())]	1✔
191	self.__json['Info']['ntot'] = sum(self.__json['Info']['ns'])	1✔
192	if self.__json['Info']['ns'][0] < 0:	1✔
193	log.warning(	1✔
194	f"The input is only in the far past the max_samples_backward is: {self.__json['Info']['ns'][0]}")
195	if self.__json['Info']['ns'][1] < 0:	1✔
196	log.warning(	1✔
197	f"The input is only in the far future the max_sample_forward is: {self.__json['Info']['ns'][1]}")
198
199	for k, v in (self.__json['Parameters'] \| self.__json['Constants']).items():	1✔
200	if 'values' in v:	1✔
201	window = 'tw' if 'tw' in v.keys() else ('sw' if 'sw' in v.keys() else None)	1✔
202	if window == 'tw':	1✔
203	check(np.array(v['values']).shape[0] == v['tw'] / self.__sample_time, ValueError,	1✔
204	f"{k} has a different number of values for this sample time.")
205	if v['values'] == "SampleTime":	1✔
206	v['values'] = self.__sample_time	1✔
207
208	def updateParameters(self, model = None, *, clear_model = False):	1✔
209	if clear_model:	1✔
210	for key in self.__json['Parameters'].keys():	1✔
211	if 'init_values' in self.__json['Parameters'][key]:	1✔
212	self.__json['Parameters'][key]['values'] = self.__json['Parameters'][key]['init_values']	1✔
213	elif 'values' in self.__json['Parameters'][key]:	1✔
214	del self.__json['Parameters'][key]['values']	1✔
215	elif model is not None:	1✔
216	for key in self.__json['Parameters'].keys():	1✔
217	if key in model.all_parameters:	1✔
218	self.__json['Parameters'][key]['values'] = model.all_parameters[key].tolist()	1✔
219

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